<s> from collections import defaultdict <EOL> import logging <EOL> import itertools <EOL> import networkx <EOL> import pyvex <EOL> import simuvex <EOL> import claripy <EOL> from archinfo import ArchARM <EOL> from . . entry_wrapper import EntryWrapper <EOL> from . . analysis import Analysis , register_analysis <EOL> from . . errors import AngrCFGError , AngrError , AngrForwardAnalysisSkipEntry <EOL> from . . knowledge import Function , FunctionManager <EOL> from . . path import make_path <EOL> from . cfg_node import CFGNode <EOL> from . cfg_base import CFGBase <EOL> from . forward_analysis import ForwardAnalysis <EOL> l = logging . getLogger ( name = "<STR_LIT>" ) <EOL> class PendingExit ( object ) : <EOL> def __init__ ( self , returning_source , state , src_simrun_key , src_exit_stmt_idx , bbl_stack , call_stack ) : <EOL> """<STR_LIT>""" <EOL> self . returning_source = returning_source <EOL> self . state = state <EOL> self . src_simrun_key = src_simrun_key <EOL> self . src_exit_stmt_idx = src_exit_stmt_idx <EOL> self . bbl_stack = bbl_stack <EOL> self . call_stack = call_stack <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % ( self . state . ip , hex ( <EOL> self . returning_source ) if self . returning_source is not None else '<STR_LIT>' ) <EOL> class CFGAccurate ( Analysis , ForwardAnalysis , CFGBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , context_sensitivity_level = <NUM_LIT:1> , <EOL> start = None , <EOL> avoid_runs = None , <EOL> enable_function_hints = False , <EOL> call_depth = None , <EOL> call_tracing_filter = None , <EOL> initial_state = None , <EOL> starts = None , <EOL> keep_state = False , <EOL> enable_advanced_backward_slicing = False , <EOL> enable_symbolic_back_traversal = False , <EOL> additional_edges = None , <EOL> no_construct = False <EOL> ) : <EOL> """<STR_LIT>""" <EOL> ForwardAnalysis . __init__ ( self ) <EOL> CFGBase . __init__ ( self , self . project , context_sensitivity_level ) <EOL> self . _symbolic_function_initial_state = { } <EOL> self . _function_input_states = None <EOL> self . _loop_back_edges_set = set ( ) <EOL> self . _unresolvable_runs = set ( ) <EOL> if start is not None : <EOL> l . warning ( "<STR_LIT>" ) <EOL> self . _starts = ( start , ) <EOL> else : <EOL> if isinstance ( starts , ( list , set ) ) : <EOL> self . _starts = tuple ( starts ) <EOL> elif isinstance ( starts , tuple ) or starts is None : <EOL> self . _starts = starts <EOL> else : <EOL> raise AngrCFGError ( '<STR_LIT>' ) <EOL> self . _avoid_runs = avoid_runs <EOL> self . _enable_function_hints = enable_function_hints <EOL> self . _call_depth = call_depth <EOL> self . _call_tracing_filter = call_tracing_filter <EOL> self . _initial_state = initial_state <EOL> self . _keep_state = keep_state <EOL> self . _advanced_backward_slicing = enable_advanced_backward_slicing <EOL> self . _enable_symbolic_back_traversal = enable_symbolic_back_traversal <EOL> self . _additional_edges = additional_edges if additional_edges else { } <EOL> self . _quasi_topological_order = { } <EOL> self . _entry_points = [ ] <EOL> self . _nodes = { } <EOL> self . _nodes_by_addr = defaultdict ( list ) <EOL> self . _sanitize_parameters ( ) <EOL> self . _executable_address_ranges = [ ] <EOL> self . _initialize_executable_ranges ( ) <EOL> if not no_construct : <EOL> self . _analyze ( ) <EOL> def copy ( self ) : <EOL> """<STR_LIT>""" <EOL> new_cfg = CFG . __new__ ( CFG ) <EOL> new_cfg . named_errors = dict ( self . named_errors ) <EOL> new_cfg . errors = list ( self . errors ) <EOL> new_cfg . _fail_fast = self . _fail_fast <EOL> new_cfg . project = self . project <EOL> new_cfg . _graph = networkx . DiGraph ( self . _graph ) <EOL> new_cfg . _nodes = self . _nodes . copy ( ) <EOL> new_cfg . _nodes_by_addr = self . _nodes_by_addr . copy ( ) <EOL> new_cfg . _edge_map = self . _edge_map . copy ( ) <EOL> new_cfg . _loop_back_edges_set = self . _loop_back_edges_set . copy ( ) <EOL> new_cfg . _loop_back_edges = self . _loop_back_edges [ : : ] <EOL> new_cfg . _executable_address_ranges = self . _executable_address_ranges [ : : ] <EOL> new_cfg . _unresolvable_runs = self . _unresolvable_runs . copy ( ) <EOL> new_cfg . _overlapped_loop_headers = self . _overlapped_loop_headers [ : : ] <EOL> new_cfg . _thumb_addrs = self . _thumb_addrs . copy ( ) <EOL> new_cfg . _keep_state = self . _keep_state <EOL> new_cfg . project = self . project <EOL> return new_cfg <EOL> def get_lbe_exits ( self ) : <EOL> """<STR_LIT>""" <EOL> for lirsb , _ in self . _loop_back_edges : <EOL> exits = lirsb . exits ( ) <EOL> yield exits <EOL> def remove_cycles ( self ) : <EOL> l . debug ( "<STR_LIT>" ) <EOL> l . debug ( "<STR_LIT>" , len ( self . _loop_back_edges ) ) <EOL> l . debug ( "<STR_LIT>" , len ( self . _overlapped_loop_headers ) ) <EOL> for b1 , b2 in self . _loop_back_edges : <EOL> if self . _graph . has_edge ( b1 , b2 ) : <EOL> l . debug ( "<STR_LIT>" , b1 , b2 ) <EOL> self . _graph . remove_edge ( b1 , b2 ) <EOL> for b in self . _overlapped_loop_headers : <EOL> successors = self . _graph . successors ( b ) <EOL> for succ in successors : <EOL> self . _graph . remove_edge ( b , succ ) <EOL> l . debug ( "<STR_LIT>" , b , succ ) <EOL> def normalize ( self ) : <EOL> """<STR_LIT>""" <EOL> graph = self . graph <EOL> end_addresses = defaultdict ( list ) <EOL> for n in graph . nodes ( ) : <EOL> if n . is_simprocedure : <EOL> continue <EOL> end_addr = n . addr + n . size <EOL> end_addresses [ ( end_addr , n . callstack_key ) ] . append ( n ) <EOL> while any ( [ len ( x ) > <NUM_LIT:1> for x in end_addresses . itervalues ( ) ] ) : <EOL> tpl_to_find = ( None , None ) <EOL> for tpl , x in end_addresses . iteritems ( ) : <EOL> if len ( x ) > <NUM_LIT:1> : <EOL> tpl_to_find = tpl <EOL> break <EOL> end_addr , callstack_key = tpl_to_find <EOL> all_nodes = end_addresses [ tpl_to_find ] <EOL> all_nodes = sorted ( all_nodes , key = lambda node : node . size ) <EOL> smallest_node = all_nodes [ <NUM_LIT:0> ] <EOL> other_nodes = all_nodes [ <NUM_LIT:1> : ] <EOL> for n in other_nodes : <EOL> new_size = smallest_node . addr - n . addr <EOL> if new_size == <NUM_LIT:0> : <EOL> continue <EOL> new_end_addr = n . addr + new_size <EOL> new_node = None <EOL> tpl = ( new_end_addr , n . callstack_key ) <EOL> if tpl in end_addresses : <EOL> nodes = [ i for i in end_addresses [ tpl ] if i . addr == n . addr ] <EOL> if len ( nodes ) > <NUM_LIT:0> : <EOL> new_node = nodes [ <NUM_LIT:0> ] <EOL> if new_node is None : <EOL> new_node = CFGNode ( n . addr , new_size , self , callstack_key = callstack_key , function_address = n . function_address ) <EOL> new_node . instruction_addrs = [ ins_addr for ins_addr in n . instruction_addrs <EOL> if ins_addr < n . addr + new_size ] <EOL> end_addresses [ tpl ] . append ( new_node ) <EOL> original_predecessors = list ( graph . in_edges_iter ( [ n ] , data = True ) ) <EOL> for p , _ , _ in original_predecessors : <EOL> graph . remove_edge ( p , n ) <EOL> graph . remove_node ( n ) <EOL> for p , _ , data in original_predecessors : <EOL> graph . add_edge ( p , new_node , data ) <EOL> new_successors = [ i for i in all_nodes <EOL> if i . addr == smallest_node . addr ] <EOL> if new_successors : <EOL> new_successor = new_successors [ <NUM_LIT:0> ] <EOL> graph . add_edge ( new_node , new_successor , jumpkind = '<STR_LIT>' ) <EOL> else : <EOL> l . error ( '<STR_LIT>' ) <EOL> end_addresses [ tpl_to_find ] = [ smallest_node ] <EOL> def downsize ( self ) : <EOL> """<STR_LIT>""" <EOL> for cfg_node in self . _nodes . itervalues ( ) : <EOL> cfg_node . downsize ( ) <EOL> def unroll_loops ( self , max_loop_unrolling_times ) : <EOL> if not isinstance ( max_loop_unrolling_times , ( int , long ) ) or max_loop_unrolling_times < <NUM_LIT:0> : <EOL> raise AngrCFGError ( '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> loop_backedges = [ ] <EOL> start = self . _starts [ <NUM_LIT:0> ] <EOL> if isinstance ( start , tuple ) : <EOL> start , _ = start <EOL> start_node = self . get_any_node ( start ) <EOL> if start_node is None : <EOL> raise AngrCFGError ( '<STR_LIT>' ) <EOL> graph_copy = networkx . DiGraph ( self . graph ) <EOL> while True : <EOL> cycles_iter = networkx . simple_cycles ( graph_copy ) <EOL> try : <EOL> cycle = cycles_iter . next ( ) <EOL> except StopIteration : <EOL> break <EOL> loop_backedge = ( None , None ) <EOL> for n in networkx . dfs_preorder_nodes ( graph_copy , source = start_node ) : <EOL> if n in cycle : <EOL> idx = cycle . index ( n ) <EOL> if idx == <NUM_LIT:0> : <EOL> loop_backedge = ( cycle [ - <NUM_LIT:1> ] , cycle [ idx ] ) <EOL> else : <EOL> loop_backedge = ( cycle [ idx - <NUM_LIT:1> ] , cycle [ idx ] ) <EOL> break <EOL> if loop_backedge not in loop_backedges : <EOL> loop_backedges . append ( loop_backedge ) <EOL> end_nodes = [ n for n in graph_copy . nodes_iter ( ) if graph_copy . out_degree ( n ) == <NUM_LIT:0> ] <EOL> new_end_node = "<STR_LIT>" <EOL> if len ( end_nodes ) == <NUM_LIT:0> : <EOL> cycles = sorted ( networkx . simple_cycles ( graph_copy ) , key = len ) <EOL> first_cycle = cycles [ <NUM_LIT:0> ] <EOL> if len ( first_cycle ) == <NUM_LIT:1> : <EOL> graph_copy . remove_edge ( first_cycle [ <NUM_LIT:0> ] , first_cycle [ <NUM_LIT:0> ] ) <EOL> else : <EOL> graph_copy . remove_edge ( first_cycle [ <NUM_LIT:0> ] , first_cycle [ <NUM_LIT:1> ] ) <EOL> end_nodes = [ n for n in graph_copy . nodes_iter ( ) if graph_copy . out_degree ( n ) == <NUM_LIT:0> ] <EOL> for en in end_nodes : <EOL> graph_copy . add_edge ( en , new_end_node ) <EOL> graph_copy . remove_node ( new_end_node ) <EOL> src , dst = loop_backedge <EOL> if graph_copy . has_edge ( src , dst ) : <EOL> new_dst = dst . copy ( ) <EOL> new_dst . looping_times = dst . looping_times + <NUM_LIT:1> <EOL> if ( <EOL> new_dst not in graph_copy and <EOL> new_dst . looping_times <= max_loop_unrolling_times ) : <EOL> dst_successors = graph_copy . successors ( dst ) <EOL> edge_data = graph_copy . get_edge_data ( src , dst ) <EOL> graph_copy . add_edge ( src , new_dst , ** edge_data ) <EOL> for ds in dst_successors : <EOL> if ds . looping_times == <NUM_LIT:0> and ds not in cycle : <EOL> edge_data = graph_copy . get_edge_data ( dst , ds ) <EOL> graph_copy . add_edge ( new_dst , ds , ** edge_data ) <EOL> graph_copy . remove_edge ( src , dst ) <EOL> self . _loop_back_edges = loop_backedges <EOL> self . _graph = graph_copy <EOL> def immediate_dominators ( self , start , target_graph = None ) : <EOL> return self . _immediate_dominators ( start , target_graph = target_graph , reverse_graph = False ) <EOL> def immediate_postdominators ( self , end , target_graph = None ) : <EOL> return self . _immediate_dominators ( end , target_graph = target_graph , reverse_graph = True ) <EOL> def remove_fakerets ( self ) : <EOL> """<STR_LIT>""" <EOL> fakeret_edges = [ ( src , dst ) for src , dst , data in self . graph . edges_iter ( data = True ) <EOL> if data [ '<STR_LIT>' ] == '<STR_LIT>' ] <EOL> self . graph . remove_edges_from ( fakeret_edges ) <EOL> def get_paths ( self , begin , end , nb_max = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> paths = self . _get_nx_paths ( begin , end ) <EOL> a_paths = [ ] <EOL> if nb_max > <NUM_LIT:0> : <EOL> paths = itertools . islice ( paths , <NUM_LIT:0> , nb_max ) <EOL> for p in paths : <EOL> runs = map ( self . irsb_from_node , p ) <EOL> a_paths . append ( make_path ( self . project , runs ) ) <EOL> return a_paths <EOL> def get_topological_order ( self , cfg_node ) : <EOL> """<STR_LIT>""" <EOL> if len ( self . _quasi_topological_order ) == <NUM_LIT:0> : <EOL> self . _quasi_topological_sort ( ) <EOL> return self . _quasi_topological_order . get ( cfg_node , None ) <EOL> def get_function_subgraph ( self , start , max_call_depth = None ) : <EOL> """<STR_LIT>""" <EOL> start_node = self . get_any_node ( start ) <EOL> node_wrapper = ( start_node , <NUM_LIT:0> ) <EOL> stack = [ node_wrapper ] <EOL> traversed_nodes = { start_node } <EOL> subgraph_nodes = set ( [ start_node ] ) <EOL> while stack : <EOL> nw = stack . pop ( ) <EOL> n , call_depth = nw [ <NUM_LIT:0> ] , nw [ <NUM_LIT:1> ] <EOL> edges = self . graph . out_edges ( n , data = True ) <EOL> for _ , dst , data in edges : <EOL> if dst not in traversed_nodes : <EOL> traversed_nodes . add ( dst ) <EOL> if data [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> if max_call_depth is None or ( max_call_depth is not None and call_depth < max_call_depth ) : <EOL> subgraph_nodes . add ( dst ) <EOL> new_nw = ( dst , call_depth + <NUM_LIT:1> ) <EOL> stack . append ( new_nw ) <EOL> elif data [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> if call_depth > <NUM_LIT:0> : <EOL> subgraph_nodes . add ( dst ) <EOL> new_nw = ( dst , call_depth - <NUM_LIT:1> ) <EOL> stack . append ( new_nw ) <EOL> else : <EOL> subgraph_nodes . add ( dst ) <EOL> new_nw = ( dst , call_depth ) <EOL> stack . append ( new_nw ) <EOL> subgraph = networkx . subgraph ( self . graph , subgraph_nodes ) <EOL> subcfg = self . copy ( ) <EOL> subcfg . _graph = subgraph <EOL> subcfg . _starts = ( start , ) <EOL> return subcfg <EOL> @ property <EOL> def functions ( self ) : <EOL> return self . kb . functions <EOL> def __setstate__ ( self , s ) : <EOL> self . project = s [ '<STR_LIT>' ] <EOL> self . _graph = s [ '<STR_LIT>' ] <EOL> self . _loop_back_edges = s [ '<STR_LIT>' ] <EOL> self . _nodes = s [ '<STR_LIT>' ] <EOL> self . _nodes_by_addr = s [ '<STR_LIT>' ] <EOL> self . _thumb_addrs = s [ '<STR_LIT>' ] <EOL> self . _unresolvable_runs = s [ '<STR_LIT>' ] <EOL> self . _executable_address_ranges = s [ '<STR_LIT>' ] <EOL> def __getstate__ ( self ) : <EOL> s = { <EOL> '<STR_LIT>' : self . project , <EOL> '<STR_LIT>' : self . _graph , <EOL> '<STR_LIT>' : self . _loop_back_edges , <EOL> '<STR_LIT>' : self . _nodes , <EOL> '<STR_LIT>' : self . _nodes_by_addr , <EOL> '<STR_LIT>' : self . _thumb_addrs , <EOL> '<STR_LIT>' : self . _unresolvable_runs , <EOL> '<STR_LIT>' : self . _executable_address_ranges , <EOL> } <EOL> return s <EOL> @ property <EOL> def unresolvables ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _unresolvable_runs <EOL> @ property <EOL> def deadends ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . graph is None : <EOL> raise AngrCFGError ( '<STR_LIT>' ) <EOL> deadends = [ i for i in self . graph if self . graph . out_degree ( i ) == <NUM_LIT:0> ] <EOL> return deadends <EOL> def _sanitize_parameters ( self ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( self . _additional_edges , ( list , set , tuple ) ) : <EOL> new_dict = defaultdict ( list ) <EOL> for s , d in self . _additional_edges : <EOL> new_dict [ s ] . append ( d ) <EOL> self . _additional_edges = new_dict <EOL> elif isinstance ( self . _additional_edges , dict ) : <EOL> pass <EOL> else : <EOL> raise AngrCFGError ( '<STR_LIT>' ) <EOL> if self . _advanced_backward_slicing and self . _enable_symbolic_back_traversal : <EOL> raise AngrCFGError ( '<STR_LIT>' ) <EOL> if self . _advanced_backward_slicing and not self . _keep_state : <EOL> raise AngrCFGError ( '<STR_LIT>' ) <EOL> self . _avoid_runs = [ ] if self . _avoid_runs is None else self . _avoid_runs <EOL> if not isinstance ( self . _avoid_runs , ( list , set ) ) : <EOL> raise AngrCFGError ( '<STR_LIT>' ) <EOL> if self . _starts is None : <EOL> self . _starts = ( ( self . project . entry , None ) , ) <EOL> else : <EOL> new_starts = [ ] <EOL> for item in self . _starts : <EOL> if isinstance ( item , tuple ) : <EOL> if len ( item ) != <NUM_LIT:2> : <EOL> raise AngrCFGError ( '<STR_LIT>' % str ( item ) ) <EOL> new_starts . append ( item ) <EOL> elif isinstance ( item , ( int , long ) ) : <EOL> new_starts . append ( ( item , None ) ) <EOL> elif isinstance ( item , simuvex . SimState ) : <EOL> new_starts . append ( item ) <EOL> self . _starts = new_starts <EOL> if not self . _starts : <EOL> raise AngrCFGError ( "<STR_LIT>" ) <EOL> def _initialize_executable_ranges ( self ) : <EOL> """<STR_LIT>""" <EOL> for b in self . project . loader . all_objects : <EOL> for seg in b . segments : <EOL> if seg . is_executable : <EOL> min_addr = seg . min_addr + b . rebase_addr <EOL> max_addr = seg . max_addr + b . rebase_addr <EOL> self . _executable_address_ranges . append ( ( min_addr , max_addr ) ) <EOL> def _init_analysis ( self ) : <EOL> """<STR_LIT>""" <EOL> ForwardAnalysis . _init_analysis ( self ) <EOL> self . _initialize_cfg ( ) <EOL> self . _function_input_states = { } <EOL> self . _loop_back_edges_set = set ( ) <EOL> self . _loop_back_edges = [ ] <EOL> self . _overlapped_loop_headers = [ ] <EOL> self . _pending_function_hints = set ( ) <EOL> self . _edge_map = defaultdict ( list ) <EOL> self . _nodes = { } <EOL> self . _nodes_by_addr = defaultdict ( list ) <EOL> self . _pending_entries = { } <EOL> self . _traced_addrs = defaultdict ( lambda : defaultdict ( int ) ) <EOL> self . _simrun_info_collection = { } <EOL> self . _analyzed_addrs = set ( ) <EOL> self . _non_returning_functions = set ( ) <EOL> for item in self . _starts : <EOL> if isinstance ( item , tuple ) : <EOL> ip = item [ <NUM_LIT:0> ] <EOL> state = self . _create_initial_state ( item [ <NUM_LIT:0> ] , item [ <NUM_LIT:1> ] ) <EOL> else : <EOL> state = item <EOL> ip = state . se . exactly_int ( state . ip ) <EOL> self . _symbolic_function_initial_state [ ip ] = state <EOL> entry_path = self . project . factory . path ( state ) <EOL> path_wrapper = EntryWrapper ( entry_path , self . _context_sensitivity_level , None , None ) <EOL> self . _entries . append ( path_wrapper ) <EOL> def _pre_analysis ( self ) : <EOL> """<STR_LIT>""" <EOL> self . kb . functions = FunctionManager ( self . kb ) <EOL> def _intra_analysis ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _pending_function_hints : <EOL> self . _process_hints ( self . _analyzed_addrs , self . _pending_entries ) <EOL> def _entry_list_empty ( self ) : <EOL> """<STR_LIT>""" <EOL> if len ( self . _pending_entries ) : <EOL> while True : <EOL> new_changes = self . _analyze_function_features ( ) <EOL> if not new_changes [ '<STR_LIT>' ] : <EOL> break <EOL> self . _clean_pending_exits ( self . _pending_entries ) <EOL> while len ( self . _pending_entries ) > <NUM_LIT:0> : <EOL> pending_entry = self . _get_pending_entry ( ) <EOL> if pending_entry is None : <EOL> continue <EOL> self . _entries . append ( pending_entry ) <EOL> break <EOL> def _create_initial_state ( self , ip , jumpkind ) : <EOL> """<STR_LIT>""" <EOL> jumpkind = "<STR_LIT>" if jumpkind is None else jumpkind <EOL> if self . _initial_state is None : <EOL> state = self . project . factory . entry_state ( addr = ip , mode = "<STR_LIT>" ) <EOL> else : <EOL> state = self . _initial_state <EOL> state . scratch . jumpkind = jumpkind <EOL> state . set_mode ( '<STR_LIT>' ) <EOL> state . ip = state . se . BVV ( ip , self . project . arch . bits ) <EOL> if jumpkind is not None : <EOL> state . scratch . jumpkind = jumpkind <EOL> state_info = None <EOL> if ip is not None and self . project . arch . name in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> state_info = { '<STR_LIT>' : state . se . BVV ( ip , self . project . arch . bits ) } <EOL> elif ip is not None and self . project . arch . name == '<STR_LIT>' : <EOL> state_info = { '<STR_LIT>' : state . registers . load ( '<STR_LIT>' ) } <EOL> state = self . project . arch . prepare_state ( state , state_info ) <EOL> return state <EOL> def _get_pending_entry ( self ) : <EOL> """<STR_LIT>""" <EOL> pending_exit_tuple = self . _pending_entries . keys ( ) [ <NUM_LIT:0> ] <EOL> pending_exit = self . _pending_entries . pop ( pending_exit_tuple ) <EOL> pending_exit_state = pending_exit . state <EOL> pending_exit_call_stack = pending_exit . call_stack <EOL> pending_exit_bbl_stack = pending_exit . bbl_stack <EOL> pending_entry_src_simrun_key = pending_exit . src_simrun_key <EOL> pending_entry_src_exit_stmt_idx = pending_exit . src_exit_stmt_idx <EOL> if pending_exit_tuple in self . _nodes : <EOL> node = self . _nodes [ pending_exit_tuple ] <EOL> if node in self . graph : <EOL> pending_exit_addr = self . _simrun_key_addr ( pending_exit_tuple ) <EOL> l . debug ( "<STR_LIT>" + "<STR_LIT>" , pending_exit_addr ) <EOL> self . _graph_add_edge ( pending_entry_src_simrun_key , pending_exit_tuple , jumpkind = "<STR_LIT>" , <EOL> exit_stmt_idx = pending_entry_src_exit_stmt_idx ) <EOL> return None <EOL> pending_exit_state . scratch . jumpkind = '<STR_LIT>' <EOL> path = self . project . factory . path ( pending_exit_state ) <EOL> entry = EntryWrapper ( path , <EOL> self . _context_sensitivity_level , <EOL> pending_entry_src_simrun_key , <EOL> pending_entry_src_exit_stmt_idx , <EOL> call_stack = pending_exit_call_stack , <EOL> bbl_stack = pending_exit_bbl_stack <EOL> ) <EOL> l . debug ( "<STR_LIT>" , self . _simrun_key_repr ( pending_exit_tuple ) ) <EOL> return entry <EOL> def _process_hints ( self , analyzed_addrs , remaining_entries ) : <EOL> """<STR_LIT>""" <EOL> while self . _pending_function_hints : <EOL> f = self . _pending_function_hints . pop ( ) <EOL> if f not in analyzed_addrs : <EOL> new_state = self . project . factory . entry_state ( '<STR_LIT>' ) <EOL> new_state . ip = new_state . se . BVV ( f , self . project . arch . bits ) <EOL> if new_state . arch . name in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> new_state . registers . store ( '<STR_LIT>' , f ) <EOL> new_path = self . project . factory . path ( new_state ) <EOL> new_path_wrapper = EntryWrapper ( new_path , <EOL> self . _context_sensitivity_level <EOL> ) <EOL> remaining_entries . append ( new_path_wrapper ) <EOL> l . debug ( '<STR_LIT>' , f ) <EOL> self . kb . functions . function ( new_path_wrapper . current_function_address , create = True ) <EOL> break <EOL> def _post_analysis ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _remove_non_return_edges ( ) <EOL> self . _analyze_calling_conventions ( ) <EOL> self . _normalize_loop_backedges ( ) <EOL> delattr ( self , "<STR_LIT>" ) <EOL> def _pre_entry_handling ( self , entry , _locals ) : <EOL> """<STR_LIT>""" <EOL> path = _locals [ '<STR_LIT:path>' ] = entry . path <EOL> call_stack_suffix = _locals [ '<STR_LIT>' ] = entry . call_stack_suffix ( ) <EOL> addr = _locals [ '<STR_LIT>' ] = entry . path . addr <EOL> func_addr = _locals [ '<STR_LIT>' ] = entry . current_function_address <EOL> _locals [ '<STR_LIT>' ] = entry . current_stack_pointer <EOL> _locals [ '<STR_LIT>' ] = entry . current_function_accessed_registers <EOL> _locals [ '<STR_LIT>' ] = self . kb . functions . function ( _locals [ '<STR_LIT>' ] , create = True ) <EOL> jumpkind = _locals [ '<STR_LIT>' ] = '<STR_LIT>' if entry . path . state . scratch . jumpkind is None else entry . path . state . scratch . jumpkind <EOL> src_simrun_key = entry . src_simrun_key <EOL> src_exit_stmt_idx = entry . src_exit_stmt_idx <EOL> if l . level == logging . DEBUG : <EOL> self . _analyzed_addrs . add ( addr ) <EOL> if addr == func_addr : <EOL> self . _function_input_states [ func_addr ] = path . state <EOL> simrun_key = self . _generate_simrun_key ( call_stack_suffix , addr , jumpkind . startswith ( '<STR_LIT>' ) ) <EOL> should_skip = False <EOL> if self . _traced_addrs [ call_stack_suffix ] [ addr ] > <NUM_LIT:1> : <EOL> should_skip = True <EOL> elif ( jumpkind == '<STR_LIT>' or jumpkind . startswith ( '<STR_LIT>' ) ) and ( self . _call_depth is not None and <EOL> len ( entry . call_stack ) > self . _call_depth and <EOL> ( <EOL> self . _call_tracing_filter is None or <EOL> self . _call_tracing_filter ( path . state , jumpkind ) <EOL> ) <EOL> ) : <EOL> should_skip = True <EOL> simrun , error_occurred , _ = self . _get_simrun ( addr , path , current_function_addr = func_addr ) <EOL> if simrun is None or should_skip : <EOL> self . _graph_add_edge ( src_simrun_key , simrun_key , jumpkind = jumpkind , exit_stmt_idx = src_exit_stmt_idx ) <EOL> self . _update_function_transition_graph ( src_simrun_key , simrun_key , jumpkind = jumpkind ) <EOL> if entry . cancelled_pending_entry is not None : <EOL> pending_entry = entry . cancelled_pending_entry <EOL> self . _graph_add_edge ( pending_entry . src_simrun_key , simrun_key , jumpkind = '<STR_LIT>' , <EOL> exit_stmt_idx = pending_entry . src_exit_stmt_idx <EOL> ) <EOL> self . _update_function_transition_graph ( pending_entry . src_simrun_key , simrun_key , jumpkind = '<STR_LIT>' ) <EOL> raise AngrForwardAnalysisSkipEntry ( ) <EOL> self . _update_thumb_addrs ( simrun , path . state ) <EOL> if self . _enable_function_hints : <EOL> function_hints = self . _search_for_function_hints ( simrun ) <EOL> for f in function_hints : <EOL> self . _pending_function_hints . add ( f ) <EOL> self . _traced_addrs [ call_stack_suffix ] [ addr ] += <NUM_LIT:1> <EOL> cfg_node = self . _create_cfgnode ( simrun , call_stack_suffix , func_addr ) <EOL> if self . _keep_state : <EOL> cfg_node . input_state = simrun . initial_state <EOL> self . _nodes [ simrun_key ] = cfg_node <EOL> self . _nodes_by_addr [ cfg_node . addr ] . append ( ( simrun_key , cfg_node ) ) <EOL> self . _graph_add_edge ( src_simrun_key , simrun_key , jumpkind = jumpkind , exit_stmt_idx = src_exit_stmt_idx ) <EOL> self . _update_function_transition_graph ( src_simrun_key , simrun_key , jumpkind = jumpkind ) <EOL> if entry . cancelled_pending_entry is not None : <EOL> pending_entry = entry . cancelled_pending_entry <EOL> self . _graph_add_edge ( pending_entry . src_simrun_key , simrun_key , jumpkind = '<STR_LIT>' , <EOL> exit_stmt_idx = pending_entry . src_exit_stmt_idx <EOL> ) <EOL> self . _update_function_transition_graph ( pending_entry . src_simrun_key , simrun_key , jumpkind = '<STR_LIT>' ) <EOL> simrun_info = self . project . arch . gather_info_from_state ( simrun . initial_state ) <EOL> self . _simrun_info_collection [ addr ] = simrun_info <EOL> _locals [ '<STR_LIT>' ] = simrun <EOL> _locals [ '<STR_LIT>' ] = simrun_key <EOL> _locals [ '<STR_LIT>' ] = cfg_node <EOL> _locals [ '<STR_LIT>' ] = error_occurred <EOL> _locals [ '<STR_LIT>' ] = { } <EOL> def _get_successors ( self , entry , _locals ) : <EOL> """<STR_LIT>""" <EOL> simrun = _locals [ '<STR_LIT>' ] <EOL> addr = _locals [ '<STR_LIT>' ] <EOL> cfg_node = _locals [ '<STR_LIT>' ] <EOL> func_addr = _locals [ '<STR_LIT>' ] <EOL> current_function = _locals [ '<STR_LIT>' ] <EOL> error_occurred = _locals [ '<STR_LIT>' ] <EOL> simrun_key = _locals [ '<STR_LIT>' ] <EOL> current_stack_pointer = _locals [ '<STR_LIT>' ] <EOL> accessed_registers_in_function = _locals [ '<STR_LIT>' ] <EOL> successors = ( simrun . flat_successors + simrun . unsat_successors ) if addr not in self . _avoid_runs else [ ] <EOL> successors , extra_info = self . _post_process_successors ( simrun , successors ) <EOL> _locals [ '<STR_LIT>' ] = extra_info <EOL> if self . _keep_state : <EOL> cfg_node . final_states = successors [ : : ] <EOL> successors = self . _resolve_indirect_jumps ( simrun , cfg_node , func_addr , successors , error_occurred , <EOL> self . _simrun_info_collection <EOL> ) <EOL> successors = self . _add_additional_edges ( simrun , cfg_node , successors ) <EOL> if len ( successors ) == <NUM_LIT:0> : <EOL> self . _push_unresolvable_run ( addr ) <EOL> if isinstance ( simrun , <EOL> simuvex . procedures . SimProcedures [ "<STR_LIT>" ] [ "<STR_LIT>" ] ) : <EOL> retn_target = entry . call_stack . current_return_target <EOL> if retn_target is not None : <EOL> new_call_stack = entry . call_stack_copy ( ) <EOL> exit_target_tpl = self . _generate_simrun_key ( <EOL> new_call_stack . stack_suffix ( self . context_sensitivity_level ) , <EOL> retn_target , <EOL> False <EOL> ) <EOL> self . _graph_add_edge ( simrun_key , exit_target_tpl , jumpkind = '<STR_LIT>' , exit_stmt_id = '<STR_LIT:default>' ) <EOL> else : <EOL> pass <EOL> if successors : <EOL> self . _handle_actions ( successors [ <NUM_LIT:0> ] , simrun , current_function , current_stack_pointer , <EOL> accessed_registers_in_function ) <EOL> return successors <EOL> def _post_entry_handling ( self , entry , successors , _locals ) : <EOL> """<STR_LIT>""" <EOL> extra_info = _locals [ '<STR_LIT>' ] <EOL> cfg_node = _locals [ '<STR_LIT>' ] <EOL> if extra_info [ '<STR_LIT>' ] and extra_info [ '<STR_LIT>' ] is not None : <EOL> cfg_node . return_target = extra_info [ '<STR_LIT>' ] <EOL> if l . level == logging . DEBUG : <EOL> self . _post_handle_entry_debug ( entry , successors , _locals ) <EOL> def _post_process_successors ( self , simrun , successors ) : <EOL> """<STR_LIT>""" <EOL> if simrun . initial_state . thumb and isinstance ( simrun , simuvex . SimIRSB ) : <EOL> pyvex . set_iropt_level ( <NUM_LIT:0> ) <EOL> it_counter = <NUM_LIT:0> <EOL> conc_temps = { } <EOL> can_produce_exits = set ( ) <EOL> bb = self . project . factory . block ( simrun . addr , thumb = True ) <EOL> for stmt in bb . vex . statements : <EOL> if stmt . tag == '<STR_LIT>' : <EOL> if it_counter > <NUM_LIT:0> : <EOL> it_counter -= <NUM_LIT:1> <EOL> can_produce_exits . add ( stmt . addr ) <EOL> elif stmt . tag == '<STR_LIT>' : <EOL> val = stmt . data <EOL> if val . tag == '<STR_LIT>' : <EOL> conc_temps [ stmt . tmp ] = val . con . val <EOL> elif stmt . tag == '<STR_LIT>' : <EOL> if stmt . offset == self . project . arch . registers [ '<STR_LIT>' ] [ <NUM_LIT:0> ] : <EOL> val = stmt . data <EOL> if val . tag == '<STR_LIT>' : <EOL> if val . tmp in conc_temps : <EOL> it_counter = <NUM_LIT:0> <EOL> itstate = conc_temps [ val . tmp ] <EOL> while itstate != <NUM_LIT:0> : <EOL> it_counter += <NUM_LIT:1> <EOL> itstate >>= <NUM_LIT:8> <EOL> if it_counter != <NUM_LIT:0> : <EOL> l . error ( '<STR_LIT>' , simrun . addr ) <EOL> THUMB_BRANCH_INSTRUCTIONS = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> for cs_insn in bb . capstone . insns : <EOL> if cs_insn . mnemonic in THUMB_BRANCH_INSTRUCTIONS : <EOL> can_produce_exits . add ( cs_insn . address ) <EOL> successors = filter ( lambda state : state . scratch . ins_addr in can_produce_exits or <EOL> state . scratch . stmt_idx == simrun . num_stmts , <EOL> successors <EOL> ) <EOL> extra_info = { '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : False , <EOL> } <EOL> for suc in successors : <EOL> suc_jumpkind = suc . scratch . jumpkind <EOL> if suc_jumpkind == "<STR_LIT>" or suc_jumpkind . startswith ( '<STR_LIT>' ) : <EOL> extra_info [ '<STR_LIT>' ] = True <EOL> if successors : <EOL> if successors [ <NUM_LIT:0> ] . scratch . jumpkind . startswith ( '<STR_LIT>' ) : <EOL> copied = successors [ <NUM_LIT:0> ] . copy ( ) <EOL> copied . scratch . jumpkind = '<STR_LIT>' <EOL> successors . append ( copied ) <EOL> return successors , extra_info <EOL> def _post_handle_entry_debug ( self , entry , successors , _locals ) : <EOL> """<STR_LIT>""" <EOL> simrun = _locals [ '<STR_LIT>' ] <EOL> call_stack_suffix = _locals [ '<STR_LIT>' ] <EOL> extra_info = _locals [ '<STR_LIT>' ] <EOL> successor_status = _locals [ '<STR_LIT>' ] <EOL> function_name = self . project . loader . find_symbol_name ( simrun . addr ) <EOL> module_name = self . project . loader . find_module_name ( simrun . addr ) <EOL> l . debug ( "<STR_LIT>" , simrun , "<STR_LIT>" . join ( [ hex ( i ) for i in call_stack_suffix if i is not None ] ) ) <EOL> l . debug ( "<STR_LIT>" , function_name , module_name ) <EOL> l . debug ( "<STR_LIT>" , extra_info [ '<STR_LIT>' ] ) <EOL> for suc in successors : <EOL> jumpkind = suc . scratch . jumpkind <EOL> if jumpkind == "<STR_LIT>" : <EOL> exit_type_str = "<STR_LIT>" <EOL> else : <EOL> exit_type_str = "<STR_LIT:->" <EOL> try : <EOL> l . debug ( "<STR_LIT>" , suc . se . exactly_int ( suc . ip ) , successor_status [ suc ] , <EOL> exit_type_str , jumpkind ) <EOL> except ( simuvex . SimValueError , simuvex . SimSolverModeError ) : <EOL> l . debug ( "<STR_LIT>" , successor_status [ suc ] , exit_type_str , <EOL> jumpkind ) <EOL> l . debug ( "<STR_LIT>" , len ( self . _entries ) , len ( self . _pending_entries ) ) <EOL> l . debug ( "<STR_LIT>" , len ( self . _analyzed_addrs ) ) <EOL> def _clean_pending_exits ( self , pending_exits ) : <EOL> """<STR_LIT>""" <EOL> pending_exits_to_remove = [ ] <EOL> for simrun_key , pe in pending_exits . iteritems ( ) : <EOL> if pe . returning_source is None : <EOL> continue <EOL> func = self . kb . functions . function ( pe . returning_source ) <EOL> if func is None : <EOL> l . warning ( "<STR_LIT>" , <EOL> hex ( pe . returning_source ) if pe . returning_source is not None else '<STR_LIT:None>' ) <EOL> continue <EOL> if func . returning is False : <EOL> pending_exits_to_remove . append ( simrun_key ) <EOL> current_function_addr = self . _simrun_key_current_func_addr ( simrun_key ) <EOL> if current_function_addr is not None : <EOL> current_function = self . kb . functions . function ( current_function_addr ) <EOL> call_site_addr = self . _simrun_key_addr ( pe . src_simrun_key ) <EOL> current_function . _call_sites [ call_site_addr ] = ( func . addr , None ) <EOL> for simrun_key in pending_exits_to_remove : <EOL> l . debug ( '<STR_LIT>' , <EOL> self . _simrun_key_addr ( simrun_key ) , <EOL> pending_exits [ simrun_key ] . returning_source , <EOL> ) <EOL> del pending_exits [ simrun_key ] <EOL> def _pre_handle_successor_state ( self , extra_info , jumpkind , target_addr ) : <EOL> """<STR_LIT>""" <EOL> if ( extra_info [ '<STR_LIT>' ] and <EOL> ( jumpkind == '<STR_LIT>' or jumpkind . startswith ( '<STR_LIT>' ) ) ) : <EOL> extra_info [ '<STR_LIT>' ] = target_addr <EOL> if ( extra_info [ '<STR_LIT>' ] and <EOL> jumpkind == '<STR_LIT>' ) : <EOL> extra_info [ '<STR_LIT>' ] = target_addr <EOL> if jumpkind == "<STR_LIT>" : <EOL> extra_info [ '<STR_LIT>' ] = target_addr <EOL> def _handle_successor ( self , entry , successor , successors , _locals ) : <EOL> """<STR_LIT>""" <EOL> state = successor <EOL> all_successor_states = successors <EOL> entry_wrapper = entry <EOL> simrun = _locals [ '<STR_LIT>' ] <EOL> simrun_key = _locals [ '<STR_LIT>' ] <EOL> addr = _locals [ '<STR_LIT>' ] <EOL> func_addr = _locals [ '<STR_LIT>' ] <EOL> call_stack_suffix = _locals [ '<STR_LIT>' ] <EOL> extra_info = _locals [ '<STR_LIT>' ] <EOL> successor_status = _locals [ '<STR_LIT>' ] <EOL> pw = None <EOL> successor_status [ state ] = "<STR_LIT>" <EOL> new_state = state . copy ( ) <EOL> suc_jumpkind = state . scratch . jumpkind <EOL> suc_exit_stmt_idx = state . scratch . exit_stmt_idx <EOL> if suc_jumpkind in { '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' } : <EOL> successor_status [ state ] = "<STR_LIT>" <EOL> return <EOL> if suc_jumpkind == "<STR_LIT>" and extra_info [ '<STR_LIT>' ] is not None : <EOL> if self . project . is_hooked ( extra_info [ '<STR_LIT>' ] ) : <EOL> sim_proc = self . project . _sim_procedures [ extra_info [ '<STR_LIT>' ] ] [ <NUM_LIT:0> ] <EOL> if sim_proc . NO_RET : <EOL> return <EOL> try : <EOL> target_addr = state . se . exactly_n_int ( state . ip , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> except ( simuvex . SimValueError , simuvex . SimSolverModeError ) : <EOL> target_addr = None <EOL> if suc_jumpkind == "<STR_LIT>" : <EOL> target_addr = entry_wrapper . call_stack . current_return_target <EOL> if target_addr is not None : <EOL> new_state . ip = new_state . se . BVV ( target_addr , new_state . arch . bits ) <EOL> if target_addr is None : <EOL> return <EOL> if state . thumb : <EOL> target_addr |= <NUM_LIT:1> <EOL> self . _pre_handle_successor_state ( extra_info , suc_jumpkind , target_addr ) <EOL> tpl = self . _generate_simrun_key ( call_stack_suffix , target_addr , suc_jumpkind . startswith ( '<STR_LIT>' ) ) <EOL> cancelled_pending_entry = None <EOL> if suc_jumpkind == '<STR_LIT>' and tpl in self . _pending_entries : <EOL> cancelled_pending_entry = self . _pending_entries [ tpl ] <EOL> l . debug ( "<STR_LIT>" , hex ( target_addr ) ) <EOL> del self . _pending_entries [ tpl ] <EOL> if suc_jumpkind == "<STR_LIT>" : <EOL> if target_addr == extra_info [ '<STR_LIT>' ] : <EOL> l . debug ( "<STR_LIT>" ) <EOL> successor_status [ state ] = "<STR_LIT>" <EOL> return <EOL> if extra_info [ '<STR_LIT>' ] : <EOL> l . debug ( '<STR_LIT>' ) <EOL> successor_status [ state ] = "<STR_LIT>" % extra_info [ '<STR_LIT>' ] <EOL> return <EOL> if ( suc_jumpkind == '<STR_LIT>' and <EOL> self . _call_depth is not None and <EOL> len ( entry_wrapper . call_stack ) <= <NUM_LIT:1> <EOL> ) : <EOL> l . debug ( '<STR_LIT>' ) <EOL> successor_status [ state ] = "<STR_LIT>" <EOL> return <EOL> new_call_stack = self . _create_new_call_stack ( addr , all_successor_states , entry_wrapper , target_addr , <EOL> suc_jumpkind ) <EOL> new_call_stack_suffix = new_call_stack . stack_suffix ( self . _context_sensitivity_level ) <EOL> new_tpl = self . _generate_simrun_key ( new_call_stack_suffix , target_addr , suc_jumpkind . startswith ( '<STR_LIT>' ) ) <EOL> if isinstance ( simrun , simuvex . SimIRSB ) : <EOL> self . _detect_loop ( simrun , <EOL> new_tpl , <EOL> simrun_key , <EOL> new_call_stack_suffix , <EOL> target_addr , <EOL> suc_jumpkind , <EOL> entry_wrapper , <EOL> func_addr ) <EOL> if suc_jumpkind == "<STR_LIT>" or suc_jumpkind . startswith ( '<STR_LIT>' ) : <EOL> new_bbl_stack = entry_wrapper . bbl_stack_copy ( ) <EOL> new_bbl_stack . call ( new_call_stack_suffix , target_addr ) <EOL> new_bbl_stack . push ( new_call_stack_suffix , target_addr , target_addr ) <EOL> elif suc_jumpkind == "<STR_LIT>" : <EOL> new_bbl_stack = entry_wrapper . bbl_stack_copy ( ) <EOL> new_bbl_stack . ret ( call_stack_suffix , func_addr ) <EOL> elif suc_jumpkind == "<STR_LIT>" : <EOL> new_bbl_stack = entry_wrapper . bbl_stack_copy ( ) <EOL> else : <EOL> new_bbl_stack = entry_wrapper . bbl_stack_copy ( ) <EOL> new_bbl_stack . push ( new_call_stack_suffix , func_addr , target_addr ) <EOL> new_path = self . project . factory . path ( new_state ) <EOL> new_path . state . set_mode ( '<STR_LIT>' ) <EOL> pw = EntryWrapper ( new_path , <EOL> self . _context_sensitivity_level , <EOL> simrun_key , <EOL> suc_exit_stmt_idx , <EOL> call_stack = new_call_stack , <EOL> bbl_stack = new_bbl_stack , <EOL> cancelled_pending_entry = cancelled_pending_entry <EOL> ) <EOL> if suc_jumpkind == "<STR_LIT>" : <EOL> if new_tpl in self . _pending_entries : <EOL> pending_entry = self . _pending_entries [ new_tpl ] <EOL> pw . cancelled_pending_entry = pending_entry <EOL> del self . _pending_entries [ new_tpl ] <EOL> successor_status [ state ] = "<STR_LIT>" <EOL> elif suc_jumpkind == "<STR_LIT>" : <EOL> st = new_state <EOL> st . set_mode ( '<STR_LIT>' ) <EOL> pw = None <EOL> pe = PendingExit ( extra_info [ '<STR_LIT>' ] , st , simrun_key , suc_exit_stmt_idx , new_bbl_stack , <EOL> new_call_stack <EOL> ) <EOL> self . _pending_entries [ new_tpl ] = pe <EOL> successor_status [ state ] = "<STR_LIT>" <EOL> elif self . _traced_addrs [ new_call_stack_suffix ] [ target_addr ] >= <NUM_LIT:1> and suc_jumpkind == "<STR_LIT>" : <EOL> pass <EOL> else : <EOL> successor_status [ state ] = "<STR_LIT>" <EOL> if extra_info [ '<STR_LIT>' ] and extra_info [ '<STR_LIT>' ] in self . _non_returning_functions : <EOL> extra_info [ '<STR_LIT>' ] = True <EOL> if pw : <EOL> self . _entries . append ( pw ) <EOL> def _handle_actions ( self , state , current_run , func , sp_addr , accessed_registers ) : <EOL> se = state . se <EOL> if func is not None and sp_addr is not None : <EOL> new_sp_addr = sp_addr + self . project . arch . call_sp_fix <EOL> actions = [ a for a in state . log . actions if a . bbl_addr == current_run . addr ] <EOL> for a in actions : <EOL> if a . type == "<STR_LIT>" and a . action == "<STR_LIT>" : <EOL> try : <EOL> addr = se . exactly_int ( a . addr . ast , default = <NUM_LIT:0> ) <EOL> except claripy . ClaripyError : <EOL> continue <EOL> if ( self . project . arch . call_pushes_ret and addr >= new_sp_addr ) or ( not self . project . arch . call_pushes_ret and addr >= new_sp_addr ) : <EOL> offset = addr - new_sp_addr <EOL> func . _add_argument_stack_variable ( offset ) <EOL> elif a . type == "<STR_LIT>" : <EOL> offset = a . offset <EOL> if a . action == "<STR_LIT>" and offset not in accessed_registers : <EOL> func . _add_argument_register ( offset ) <EOL> elif a . action == "<STR_LIT>" : <EOL> accessed_registers . add ( offset ) <EOL> else : <EOL> l . error ( <EOL> "<STR_LIT>" ) <EOL> def _graph_get_node ( self , node_key , terminator_for_nonexistent_node = False ) : <EOL> """<STR_LIT>""" <EOL> if node_key not in self . _nodes : <EOL> if not terminator_for_nonexistent_node : <EOL> return None <EOL> addr = self . _simrun_key_addr ( node_key ) <EOL> func_addr = self . _simrun_key_current_func_addr ( node_key ) <EOL> if func_addr is None : <EOL> func_addr = self . _simrun_key_addr ( node_key ) <EOL> pt = CFGNode ( self . _simrun_key_addr ( node_key ) , <EOL> None , <EOL> self , <EOL> callstack_key = self . _simrun_key_callstack_key ( node_key ) , <EOL> input_state = None , <EOL> simprocedure_name = "<STR_LIT>" , <EOL> function_address = func_addr ) <EOL> if self . _keep_state : <EOL> pt . input_state = self . project . factory . entry_state ( ) <EOL> pt . input_state . ip = pt . addr <EOL> self . _nodes [ node_key ] = pt <EOL> self . _nodes_by_addr [ pt . addr ] . append ( ( node_key , pt ) ) <EOL> if isinstance ( self . project . arch , ArchARM ) and addr % <NUM_LIT:2> == <NUM_LIT:1> : <EOL> self . _thumb_addrs . add ( addr ) <EOL> self . _thumb_addrs . add ( addr - <NUM_LIT:1> ) <EOL> l . debug ( "<STR_LIT>" , <EOL> self . _simrun_key_repr ( node_key ) ) <EOL> return self . _nodes [ node_key ] <EOL> def _graph_add_edge ( self , src_node_key , dst_node_key , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> dst_node = self . _graph_get_node ( dst_node_key , terminator_for_nonexistent_node = True ) <EOL> if src_node_key is None : <EOL> self . graph . add_node ( dst_node ) <EOL> else : <EOL> src_node = self . _graph_get_node ( src_node_key , terminator_for_nonexistent_node = True ) <EOL> self . graph . add_edge ( src_node , dst_node , ** kwargs ) <EOL> def _update_function_transition_graph ( self , src_node_key , dst_node_key , jumpkind = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> dst_node = self . _graph_get_node ( dst_node_key , terminator_for_nonexistent_node = True ) <EOL> if src_node_key is None : <EOL> self . kb . functions . function ( dst_node . function_address , create = True ) . _register_nodes ( dst_node . to_codenode ( ) ) <EOL> return <EOL> src_node = self . _graph_get_node ( src_node_key , terminator_for_nonexistent_node = True ) <EOL> if jumpkind == "<STR_LIT>" : <EOL> ret_addr = src_node . return_target <EOL> ret_node = self . kb . functions . function ( <EOL> src_node . function_address , <EOL> create = True <EOL> ) . _get_block ( ret_addr ) . codenode if ret_addr else None <EOL> if ret_node is None : <EOL> l . warning ( "<STR_LIT>" , dst_node . addr , src_node . addr ) <EOL> self . kb . functions . _add_call_to ( <EOL> function_addr = src_node . function_address , <EOL> from_node = src_node . to_codenode ( ) , <EOL> to_addr = dst_node . addr , <EOL> retn_node = ret_node , <EOL> syscall = False <EOL> ) <EOL> if jumpkind . startswith ( '<STR_LIT>' ) : <EOL> self . kb . functions . _add_call_to ( <EOL> function_addr = src_node . function_address , <EOL> from_node = src_node . to_codenode ( ) , <EOL> to_addr = dst_node . addr , <EOL> retn_node = src_node . to_codenode ( ) , <EOL> syscall = True , <EOL> ) <EOL> elif jumpkind == '<STR_LIT>' : <EOL> self . kb . functions . _add_return_from ( <EOL> function_addr = src_node . function_address , <EOL> from_node = src_node . to_codenode ( ) , <EOL> to_node = dst_node . to_codenode ( ) , <EOL> ) <EOL> self . kb . functions . _add_return_from_call ( <EOL> function_addr = dst_node . function_address , <EOL> src_function_addr = src_node . function_address , <EOL> to_node = dst_node . to_codenode ( ) <EOL> ) <EOL> elif jumpkind == '<STR_LIT>' : <EOL> self . kb . functions . _add_fakeret_to ( <EOL> function_addr = src_node . function_address , <EOL> from_node = src_node . to_codenode ( ) , <EOL> to_node = dst_node . to_codenode ( ) <EOL> ) <EOL> elif jumpkind == '<STR_LIT>' : <EOL> src_obj = self . project . loader . addr_belongs_to_object ( src_node . addr ) <EOL> dest_obj = self . project . loader . addr_belongs_to_object ( dst_node . addr ) <EOL> if src_obj is dest_obj : <EOL> self . kb . functions . _add_transition_to ( <EOL> function_addr = src_node . function_address , <EOL> from_node = src_node . to_codenode ( ) , <EOL> to_node = dst_node . to_codenode ( ) <EOL> ) <EOL> else : <EOL> self . kb . functions . _add_call_to ( <EOL> function_addr = src_node . function_address , <EOL> from_node = src_node . to_codenode ( ) , <EOL> to_addr = dst_node . addr , <EOL> retn_node = None , <EOL> syscall = False <EOL> ) <EOL> def _add_additional_edges ( self , simrun , cfg_node , successors ) : <EOL> """<STR_LIT>""" <EOL> addr = cfg_node . addr <EOL> if addr in self . _additional_edges : <EOL> dests = self . _additional_edges [ addr ] <EOL> for dst in dests : <EOL> if isinstance ( simrun , simuvex . SimIRSB ) : <EOL> base_state = simrun . default_exit . copy ( ) <EOL> else : <EOL> if successors : <EOL> base_state = successors [ <NUM_LIT:0> ] . copy ( ) <EOL> else : <EOL> base_state = simrun . initial_state <EOL> base_state . ip = dst <EOL> successors . append ( base_state ) <EOL> l . debug ( "<STR_LIT>" , dst , simrun ) <EOL> return successors <EOL> def _remove_non_return_edges ( self ) : <EOL> """<STR_LIT>""" <EOL> for func in self . kb . functions . values ( ) : <EOL> graph = func . transition_graph <EOL> all_return_edges = [ ( u , v ) for ( u , v , data ) in graph . edges ( data = True ) if data [ '<STR_LIT:type>' ] == '<STR_LIT>' ] <EOL> for return_from_call_edge in all_return_edges : <EOL> callsite_block_addr , return_to_addr = return_from_call_edge <EOL> call_func_addr = func . get_call_target ( callsite_block_addr ) <EOL> if call_func_addr is None : <EOL> continue <EOL> call_func = self . kb . functions . function ( call_func_addr ) <EOL> if call_func is None : <EOL> continue <EOL> if call_func . returning is False : <EOL> graph . remove_edge ( call_func_addr , return_to_addr ) <EOL> nodes = self . get_all_nodes ( callsite_block_addr ) <EOL> for n in nodes : <EOL> successors = self . get_successors_and_jumpkind ( n , excluding_fakeret = False ) <EOL> for successor , jumpkind in successors : <EOL> if jumpkind == '<STR_LIT>' and successor . addr == return_to_addr : <EOL> self . remove_edge ( n , successor ) <EOL> def _resolve_indirect_jumps ( self , simrun , cfg_node , func_addr , successors , error_occurred , simrun_info_collection ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( simrun , simuvex . SimIRSB ) and self . _is_indirect_jump ( cfg_node , simrun ) : <EOL> l . debug ( '<STR_LIT>' , simrun . addr ) <EOL> old_successors = successors [ : : ] <EOL> successors = [ ] <EOL> for i , suc in enumerate ( old_successors ) : <EOL> if suc . se . symbolic ( suc . ip ) : <EOL> successors . append ( suc ) <EOL> else : <EOL> ip_int = suc . se . exactly_int ( suc . ip ) <EOL> if self . _is_address_executable ( ip_int ) or self . project . is_hooked ( ip_int ) : <EOL> successors . append ( suc ) <EOL> else : <EOL> l . info ( '<STR_LIT>' , <EOL> cfg_node , <EOL> i + <NUM_LIT:1> , len ( old_successors ) , <EOL> ip_int ) <EOL> if self . _advanced_backward_slicing and self . _keep_state : <EOL> should_resolve = True <EOL> legit_successors = [ suc for suc in successors if suc . scratch . jumpkind in ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> if legit_successors : <EOL> legit_successor = legit_successors [ <NUM_LIT:0> ] <EOL> if legit_successor . ip . symbolic : <EOL> if not legit_successor . scratch . jumpkind == '<STR_LIT>' : <EOL> should_resolve = False <EOL> else : <EOL> if legit_successor . scratch . jumpkind == '<STR_LIT>' : <EOL> should_resolve = False <EOL> else : <EOL> concrete_target = legit_successor . se . any_int ( legit_successor . ip ) <EOL> if not self . project . loader . addr_belongs_to_object ( <EOL> concrete_target ) is self . project . loader . main_bin : <EOL> should_resolve = False <EOL> else : <EOL> should_resolve = False <EOL> if not should_resolve : <EOL> l . debug ( "<STR_LIT>" ) <EOL> else : <EOL> more_successors = self . _resolve_indirect_jump ( cfg_node , simrun , func_addr ) <EOL> if len ( more_successors ) : <EOL> all_successors = [ suc for suc in successors if not suc . se . symbolic ( suc . ip ) ] <EOL> for suc_addr in more_successors : <EOL> a = simrun . default_exit . copy ( ) <EOL> a . ip = suc_addr <EOL> all_successors . insert ( <NUM_LIT:0> , a ) <EOL> l . debug ( '<STR_LIT>' ) <EOL> self . kb . _resolved_indirect_jumps . add ( simrun . addr ) <EOL> else : <EOL> l . debug ( '<STR_LIT>' ) <EOL> self . kb . _unresolved_indirect_jumps . add ( simrun . addr ) <EOL> else : <EOL> if not successors : <EOL> l . debug ( '<STR_LIT>' , <EOL> cfg_node ) <EOL> if not error_occurred and ( cfg_node . is_simprocedure or self . _is_indirect_jump ( cfg_node , simrun ) ) : <EOL> has_call_jumps = any ( [ suc_state . scratch . jumpkind == '<STR_LIT>' for suc_state in successors ] ) <EOL> if has_call_jumps : <EOL> concrete_successors = [ suc_state for suc_state in successors if <EOL> suc_state . scratch . jumpkind != '<STR_LIT>' and not suc_state . se . symbolic ( <EOL> suc_state . ip ) ] <EOL> else : <EOL> concrete_successors = [ suc_state for suc_state in successors if <EOL> not suc_state . se . symbolic ( suc_state . ip ) ] <EOL> symbolic_successors = [ suc_state for suc_state in successors if suc_state . se . symbolic ( suc_state . ip ) ] <EOL> resolved = len ( symbolic_successors ) == <NUM_LIT:0> <EOL> if len ( symbolic_successors ) > <NUM_LIT:0> : <EOL> for suc in symbolic_successors : <EOL> if simuvex . o . SYMBOLIC in suc . options : <EOL> targets = suc . se . any_n_int ( suc . ip , <NUM_LIT:32> ) <EOL> if len ( targets ) < <NUM_LIT:32> : <EOL> all_successors = [ ] <EOL> resolved = True <EOL> for t in targets : <EOL> new_ex = suc . copy ( ) <EOL> new_ex . ip = suc . se . BVV ( t , suc . ip . size ( ) ) <EOL> all_successors . append ( new_ex ) <EOL> else : <EOL> break <EOL> if not resolved and ( <EOL> ( len ( symbolic_successors ) > <NUM_LIT:0> and len ( concrete_successors ) == <NUM_LIT:0> ) or <EOL> ( not cfg_node . is_simprocedure and self . _is_indirect_jump ( cfg_node , simrun ) ) <EOL> ) : <EOL> l . debug ( "<STR_LIT>" , cfg_node ) <EOL> if isinstance ( simrun , simuvex . SimProcedure ) and simrun . ADDS_EXITS : <EOL> l . debug ( '<STR_LIT>' , simrun ) <EOL> if self . _enable_symbolic_back_traversal : <EOL> successors = self . _symbolically_back_traverse ( simrun , simrun_info_collection , cfg_node ) <EOL> if len ( successors ) : <EOL> self . kb . _resolved_indirect_jumps . add ( simrun . addr ) <EOL> else : <EOL> self . kb . _unresolved_indirect_jumps . add ( simrun . addr ) <EOL> l . debug ( "<STR_LIT>" , len ( successors ) ) <EOL> else : <EOL> self . kb . _unresolved_indirect_jumps . add ( simrun . addr ) <EOL> successors = [ s for s in successors if s . scratch . jumpkind == "<STR_LIT>" ] <EOL> elif isinstance ( simrun , simuvex . SimIRSB ) and any ( [ ex . scratch . jumpkind != '<STR_LIT>' for ex in successors ] ) : <EOL> l . debug ( "<STR_LIT>" , cfg_node ) <EOL> if self . _enable_symbolic_back_traversal : <EOL> successors = self . _symbolically_back_traverse ( simrun , simrun_info_collection , cfg_node ) <EOL> successors = [ suc for suc in successors <EOL> if self . _is_address_executable ( suc . se . exactly_int ( suc . ip ) ) ] <EOL> if len ( successors ) : <EOL> self . kb . _resolved_indirect_jumps . add ( simrun . addr ) <EOL> else : <EOL> self . kb . _unresolved_indirect_jumps . add ( simrun . addr ) <EOL> l . debug ( '<STR_LIT>' , len ( successors ) ) <EOL> else : <EOL> self . kb . _unresolved_indirect_jumps . add ( simrun . addr ) <EOL> successors = [ ] <EOL> elif len ( successors ) > <NUM_LIT:0> and all ( [ ex . scratch . jumpkind == '<STR_LIT>' for ex in successors ] ) : <EOL> l . debug ( '<STR_LIT>' ) <EOL> else : <EOL> l . warning ( '<STR_LIT>' , cfg_node ) <EOL> self . kb . _unresolved_indirect_jumps . add ( simrun . addr ) <EOL> return successors <EOL> def _resolve_indirect_jump ( self , cfgnode , simirsb , current_function_addr ) : <EOL> """<STR_LIT>""" <EOL> l . debug ( "<STR_LIT>" , simirsb ) <EOL> next_tmp = simirsb . irsb . next . tmp <EOL> stmt_id = [ i for i , s in enumerate ( simirsb . irsb . statements ) <EOL> if isinstance ( s , pyvex . IRStmt . WrTmp ) and s . tmp == next_tmp ] [ <NUM_LIT:0> ] <EOL> cdg = self . project . analyses . CDG ( cfg = self ) <EOL> ddg = self . project . analyses . DDG ( cfg = self , start = current_function_addr , call_depth = <NUM_LIT:0> ) <EOL> bc = self . project . analyses . BackwardSlice ( self , cdg , ddg , targets = [ ( cfgnode , stmt_id ) ] , same_function = True ) <EOL> taint_graph = bc . taint_graph <EOL> next_nodes = [ cl for cl in taint_graph . nodes ( ) if cl . simrun_addr == simirsb . addr ] <EOL> if not next_nodes : <EOL> l . error ( '<STR_LIT>' ) <EOL> return [ ] <EOL> next_node = next_nodes [ <NUM_LIT:0> ] <EOL> all_subgraphs = networkx . weakly_connected_component_subgraphs ( taint_graph ) <EOL> starts = set ( ) <EOL> for subgraph in all_subgraphs : <EOL> if next_node in subgraph : <EOL> nodes = [ n for n in subgraph . nodes ( ) if subgraph . in_degree ( n ) == <NUM_LIT:0> ] <EOL> for n in nodes : <EOL> starts . add ( n . simrun_addr ) <EOL> successing_addresses = set ( ) <EOL> annotated_cfg = bc . annotated_cfg ( ) <EOL> for start in starts : <EOL> l . debug ( '<STR_LIT>' , start ) <EOL> node = self . get_any_node ( start ) <EOL> if node is None : <EOL> p = self . project . factory . path ( self . project . factory . blank_state ( addr = start ) ) <EOL> else : <EOL> base_state = node . input_state . copy ( ) <EOL> base_state . set_mode ( '<STR_LIT>' ) <EOL> base_state . ip = start <EOL> initial_nodes = [ n for n in bc . taint_graph . nodes ( ) if bc . taint_graph . in_degree ( n ) == <NUM_LIT:0> ] <EOL> for cl in initial_nodes : <EOL> cfg_nodes = self . get_all_nodes ( cl . simrun_addr ) <EOL> for n in cfg_nodes : <EOL> if not n . final_states : <EOL> continue <EOL> actions = [ ac for ac in n . final_states [ <NUM_LIT:0> ] . log . actions <EOL> if ac . bbl_addr == cl . simrun_addr and <EOL> ac . stmt_idx == cl . stmt_idx <EOL> ] <EOL> for ac in actions : <EOL> if not hasattr ( ac , '<STR_LIT:action>' ) : <EOL> continue <EOL> if ac . action == '<STR_LIT>' : <EOL> if ac . type == '<STR_LIT>' : <EOL> unconstrained_value = base_state . se . Unconstrained ( '<STR_LIT>' , <EOL> ac . size . ast * <NUM_LIT:8> ) <EOL> base_state . memory . store ( ac . addr , <EOL> unconstrained_value , <EOL> endness = self . project . arch . memory_endness ) <EOL> elif ac . type == '<STR_LIT>' : <EOL> unconstrained_value = base_state . se . Unconstrained ( '<STR_LIT>' , <EOL> ac . size . ast * <NUM_LIT:8> ) <EOL> base_state . registers . store ( ac . offset , <EOL> unconstrained_value , <EOL> endness = self . project . arch . register_endness ) <EOL> base_state . release_plugin ( '<STR_LIT>' ) <EOL> p = self . project . factory . path ( base_state ) <EOL> old_timeout = p . state . se . _solver . timeout <EOL> p . state . se . _solver . timeout = <NUM_LIT> <EOL> sc = self . project . surveyors . Slicecutor ( annotated_cfg , start = p , max_loop_iterations = <NUM_LIT:1> ) . run ( ) <EOL> p . state . se . _solver . timeout = old_timeout <EOL> if sc . cut or sc . deadended : <EOL> all_deadended_paths = sc . cut + sc . deadended <EOL> for p in all_deadended_paths : <EOL> if p . addr == simirsb . addr : <EOL> successing_paths = p . step ( ) <EOL> for sp in successing_paths : <EOL> successing_addresses . add ( sp . addr ) <EOL> else : <EOL> l . debug ( "<STR_LIT>" ) <EOL> l . debug ( '<STR_LIT>' , len ( successing_addresses ) ) <EOL> return list ( successing_addresses ) <EOL> def _symbolically_back_traverse ( self , current_simrun , simrun_info_collection , cfg_node ) : <EOL> class register_protector ( object ) : <EOL> def __init__ ( self , reg_offset , info_collection ) : <EOL> self . _reg_offset = reg_offset <EOL> self . _info_collection = info_collection <EOL> def write_persistent_register ( self , state_ ) : <EOL> if state_ . inspect . address is None : <EOL> l . error ( '<STR_LIT>' ) <EOL> return <EOL> if state_ . registers . load ( self . _reg_offset ) . symbolic : <EOL> current_run = state_ . inspect . address <EOL> if current_run in self . _info_collection and not state_ . se . symbolic ( self . _info_collection [ current_run ] [ self . _reg_offset ] ) : <EOL> l . debug ( "<STR_LIT>" , state_ . registers . load ( self . _reg_offset ) , <EOL> self . _info_collection [ current_run ] [ self . _reg_offset ] ) <EOL> state_ . registers . store ( <EOL> self . _reg_offset , <EOL> self . _info_collection [ current_run ] [ self . _reg_offset ] <EOL> ) <EOL> l . debug ( "<STR_LIT>" , current_simrun ) <EOL> temp_cfg = networkx . DiGraph ( self . graph ) <EOL> temp_cfg . reverse ( copy = False ) <EOL> path_length = <NUM_LIT:0> <EOL> concrete_exits = [ ] <EOL> if cfg_node not in temp_cfg . nodes ( ) : <EOL> return concrete_exits <EOL> keep_running = True <EOL> while len ( concrete_exits ) == <NUM_LIT:0> and path_length < <NUM_LIT:5> and keep_running : <EOL> path_length += <NUM_LIT:1> <EOL> queue = [ cfg_node ] <EOL> avoid = set ( ) <EOL> for _ in xrange ( path_length ) : <EOL> new_queue = [ ] <EOL> for n in queue : <EOL> successors = temp_cfg . successors ( n ) <EOL> for suc in successors : <EOL> jk = temp_cfg . get_edge_data ( n , suc ) [ '<STR_LIT>' ] <EOL> if jk != '<STR_LIT>' : <EOL> predecessors = temp_cfg . predecessors ( suc ) <EOL> avoid |= set ( [ p . addr for p in predecessors if p is not n ] ) <EOL> new_queue . append ( suc ) <EOL> queue = new_queue <EOL> if path_length <= <NUM_LIT:1> : <EOL> continue <EOL> for n in queue : <EOL> state = self . project . factory . blank_state ( addr = n . addr , <EOL> mode = '<STR_LIT>' , <EOL> add_options = { <EOL> simuvex . o . DO_RET_EMULATION , <EOL> simuvex . o . CONSERVATIVE_READ_STRATEGY , <EOL> } | simuvex . o . resilience_options <EOL> ) <EOL> if n . addr in simrun_info_collection : <EOL> for reg in state . arch . persistent_regs : <EOL> state . registers . store ( reg , simrun_info_collection [ n . addr ] [ reg ] ) <EOL> for reg in state . arch . persistent_regs : <EOL> reg_protector = register_protector ( reg , simrun_info_collection ) <EOL> state . inspect . add_breakpoint ( '<STR_LIT>' , <EOL> simuvex . BP ( <EOL> simuvex . BP_AFTER , <EOL> reg_write_offset = state . arch . registers [ reg ] [ <NUM_LIT:0> ] , <EOL> action = reg_protector . write_persistent_register <EOL> ) <EOL> ) <EOL> result = self . project . surveyors . Explorer ( <EOL> start = self . project . factory . path ( state ) , <EOL> find = ( current_simrun . addr , ) , <EOL> avoid = avoid , <EOL> max_repeats = <NUM_LIT:10> , <EOL> max_depth = path_length <EOL> ) . run ( ) <EOL> if result . found : <EOL> if not result . found [ <NUM_LIT:0> ] . errored and len ( result . found [ <NUM_LIT:0> ] . step ( ) ) > <NUM_LIT:0> : <EOL> keep_running = False <EOL> concrete_exits . extend ( [ s for s in result . found [ <NUM_LIT:0> ] . next_run . flat_successors ] ) <EOL> concrete_exits . extend ( [ s for s in result . found [ <NUM_LIT:0> ] . next_run . unsat_successors ] ) <EOL> if keep_running : <EOL> l . debug ( '<STR_LIT>' ) <EOL> new_concrete_successors = [ ] <EOL> for c in concrete_exits : <EOL> unsat_state = current_simrun . unsat_successors [ <NUM_LIT:0> ] . copy ( ) <EOL> unsat_state . scratch . jumpkind = c . scratch . jumpkind <EOL> for reg in unsat_state . arch . persistent_regs + [ '<STR_LIT>' ] : <EOL> unsat_state . registers . store ( reg , c . registers . load ( reg ) ) <EOL> new_concrete_successors . append ( unsat_state ) <EOL> return new_concrete_successors <EOL> def _get_symbolic_function_initial_state ( self , function_addr , fastpath_mode_state = None ) : <EOL> """<STR_LIT>""" <EOL> if function_addr is None : <EOL> return None <EOL> if function_addr in self . _symbolic_function_initial_state : <EOL> return self . _symbolic_function_initial_state [ function_addr ] <EOL> if fastpath_mode_state is not None : <EOL> fastpath_state = fastpath_mode_state <EOL> else : <EOL> if function_addr in self . _function_input_states : <EOL> fastpath_state = self . _function_input_states [ function_addr ] <EOL> else : <EOL> raise AngrCFGError ( '<STR_LIT>' ) <EOL> symbolic_initial_state = self . project . factory . entry_state ( mode = '<STR_LIT>' ) <EOL> if fastpath_state is not None : <EOL> symbolic_initial_state = self . project . _simos . prepare_call_state ( fastpath_state , <EOL> initial_state = symbolic_initial_state ) <EOL> try : <EOL> tmp_block = self . project . factory . block ( function_addr ) <EOL> except ( simuvex . SimError , AngrError ) : <EOL> return None <EOL> num_instr = tmp_block . instructions - <NUM_LIT:1> <EOL> symbolic_initial_state . ip = function_addr <EOL> path = self . project . factory . path ( symbolic_initial_state ) <EOL> try : <EOL> simrun = self . project . factory . sim_run ( path . state , num_inst = num_instr ) <EOL> except ( simuvex . SimError , AngrError ) : <EOL> return None <EOL> exits = simrun . flat_successors + simrun . unsat_successors <EOL> if exits : <EOL> final_st = None <EOL> for ex in exits : <EOL> if ex . satisfiable ( ) : <EOL> final_st = ex <EOL> break <EOL> else : <EOL> final_st = None <EOL> self . _symbolic_function_initial_state [ function_addr ] = final_st <EOL> return final_st <EOL> def _search_for_function_hints ( self , simrun ) : <EOL> """<STR_LIT>""" <EOL> function_hints = [ ] <EOL> if isinstance ( simrun , simuvex . SimIRSB ) and simrun . successors : <EOL> successor = simrun . successors [ <NUM_LIT:0> ] <EOL> for action in successor . log . actions : <EOL> if action . type == '<STR_LIT>' and action . offset == self . project . arch . ip_offset : <EOL> continue <EOL> elif action . type == '<STR_LIT>' : <EOL> continue <EOL> data = action . data <EOL> if data is not None : <EOL> try : <EOL> const = successor . se . exactly_n_int ( data . ast , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> except : <EOL> continue <EOL> if self . _is_address_executable ( const ) : <EOL> if self . _pending_function_hints is not None and const in self . _pending_function_hints : <EOL> continue <EOL> function_hints . append ( const ) <EOL> l . info ( '<STR_LIT>' , len ( function_hints ) , simrun , <EOL> "<STR_LIT:U+002CU+0020>" . join ( [ "<STR_LIT>" % f for f in function_hints ] ) ) <EOL> return function_hints <EOL> def _get_simrun ( self , addr , current_entry , current_function_addr = None ) : <EOL> error_occurred = False <EOL> state = current_entry . state <EOL> saved_state = current_entry . state <EOL> try : <EOL> if not self . _keep_state and self . project . is_hooked ( addr ) and not self . project . _sim_procedures [ addr ] [ <NUM_LIT:0> ] is simuvex . s_procedure . SimProcedureContinuation and not self . project . _sim_procedures [ addr ] [ <NUM_LIT:0> ] . ADDS_EXITS and not self . project . _sim_procedures [ addr ] [ <NUM_LIT:0> ] . NO_RET : <EOL> old_proc = self . project . _sim_procedures [ addr ] [ <NUM_LIT:0> ] <EOL> old_name = None <EOL> if old_proc == simuvex . procedures . SimProcedures [ "<STR_LIT>" ] [ "<STR_LIT>" ] : <EOL> proc_kwargs = self . project . _sim_procedures [ addr ] [ <NUM_LIT:1> ] <EOL> if '<STR_LIT>' in proc_kwargs : <EOL> old_name = proc_kwargs [ '<STR_LIT>' ] <EOL> if old_name is None : <EOL> old_name = old_proc . __name__ . split ( '<STR_LIT:.>' ) [ - <NUM_LIT:1> ] <EOL> sim_run = simuvex . procedures . SimProcedures [ "<STR_LIT>" ] [ "<STR_LIT>" ] ( <EOL> state , <EOL> addr = addr , <EOL> sim_kwargs = { '<STR_LIT>' : "<STR_LIT:%s>" % old_name } <EOL> ) <EOL> else : <EOL> jumpkind = state . scratch . jumpkind <EOL> jumpkind = '<STR_LIT>' if jumpkind is None else jumpkind <EOL> sim_run = self . project . factory . sim_run ( current_entry . state , jumpkind = jumpkind ) <EOL> except ( simuvex . SimFastPathError , simuvex . SimSolverModeError ) as ex : <EOL> if saved_state . mode == '<STR_LIT>' : <EOL> l . debug ( '<STR_LIT>' , addr ) <EOL> l . debug ( '<STR_LIT>' % addr ) <EOL> new_state = None <EOL> if addr != current_function_addr : <EOL> new_state = self . _get_symbolic_function_initial_state ( current_function_addr ) <EOL> if new_state is None : <EOL> new_state = current_entry . state . copy ( ) <EOL> new_state . set_mode ( '<STR_LIT>' ) <EOL> new_state . options . add ( simuvex . o . DO_RET_EMULATION ) <EOL> new_state . se . _solver . constraints = [ c for c in new_state . se . constraints if <EOL> c . op != '<STR_LIT:I>' or c . args [ <NUM_LIT:0> ] is not False ] <EOL> new_state . se . _solver . _result = None <EOL> saved_state , current_entry . state = current_entry . state , new_state <EOL> sim_run , error_occurred , _ = self . _get_simrun ( addr , current_entry ) <EOL> else : <EOL> l . debug ( "<STR_LIT>" , ex ) <EOL> error_occurred = True <EOL> sim_run = simuvex . procedures . SimProcedures [ "<STR_LIT>" ] [ "<STR_LIT>" ] ( <EOL> state , addr = addr ) <EOL> except simuvex . SimIRSBError : <EOL> l . debug ( "<STR_LIT>" , exc_info = True ) <EOL> error_occurred = True <EOL> sim_run = simuvex . procedures . SimProcedures [ "<STR_LIT>" ] [ "<STR_LIT>" ] ( <EOL> state , addr = addr ) <EOL> except claripy . ClaripyError : <EOL> l . debug ( "<STR_LIT>" , exc_info = True ) <EOL> error_occurred = True <EOL> sim_run = simuvex . procedures . SimProcedures [ "<STR_LIT>" ] [ "<STR_LIT>" ] ( <EOL> state , addr = addr ) <EOL> except simuvex . SimError : <EOL> l . debug ( "<STR_LIT>" , exc_info = True ) <EOL> error_occurred = True <EOL> sim_run = simuvex . procedures . SimProcedures [ "<STR_LIT>" ] [ "<STR_LIT>" ] ( <EOL> state , addr = addr ) <EOL> except AngrError : <EOL> section = self . project . loader . main_bin . find_section_containing ( addr ) <EOL> if section is None : <EOL> sec_name = '<STR_LIT>' <EOL> else : <EOL> sec_name = section . name <EOL> l . error ( "<STR_LIT>" , <EOL> addr , sec_name , exc_info = True ) <EOL> error_occurred = True <EOL> sim_run = None <EOL> return sim_run , error_occurred , saved_state <EOL> def _create_new_call_stack ( self , addr , all_entries , entry_wrapper , exit_target , jumpkind ) : <EOL> if jumpkind == "<STR_LIT>" or jumpkind . startswith ( '<STR_LIT>' ) : <EOL> new_call_stack = entry_wrapper . call_stack_copy ( ) <EOL> if len ( all_entries ) > <NUM_LIT:1> and all_entries [ - <NUM_LIT:1> ] . scratch . jumpkind == "<STR_LIT>" : <EOL> se = all_entries [ - <NUM_LIT:1> ] . se <EOL> retn_target_addr = se . exactly_int ( all_entries [ - <NUM_LIT:1> ] . ip , default = <NUM_LIT:0> ) <EOL> sp = se . exactly_int ( all_entries [ - <NUM_LIT:1> ] . regs . sp , default = <NUM_LIT:0> ) <EOL> new_call_stack . call ( addr , exit_target , <EOL> retn_target = retn_target_addr , <EOL> stack_pointer = sp ) <EOL> elif jumpkind . startswith ( '<STR_LIT>' ) and len ( all_entries ) == <NUM_LIT:1> : <EOL> retn_target_addr = exit_target <EOL> se = all_entries [ <NUM_LIT:0> ] . se <EOL> sp = se . exactly_int ( all_entries [ <NUM_LIT:0> ] . regs . sp , default = <NUM_LIT:0> ) <EOL> new_call_stack . call ( addr , exit_target , <EOL> retn_target = retn_target_addr , <EOL> stack_pointer = sp ) <EOL> else : <EOL> new_call_stack . clear ( ) <EOL> se = all_entries [ - <NUM_LIT:1> ] . se <EOL> sp = se . exactly_int ( all_entries [ - <NUM_LIT:1> ] . regs . sp , default = <NUM_LIT:0> ) <EOL> new_call_stack . call ( addr , exit_target , retn_target = None , stack_pointer = sp ) <EOL> elif jumpkind == "<STR_LIT>" : <EOL> new_call_stack = entry_wrapper . call_stack_copy ( ) <EOL> new_call_stack . ret ( exit_target ) <EOL> se = all_entries [ - <NUM_LIT:1> ] . se <EOL> sp = se . exactly_int ( all_entries [ - <NUM_LIT:1> ] . regs . sp , default = <NUM_LIT:0> ) <EOL> old_sp = entry_wrapper . current_stack_pointer <EOL> if sp is not None and old_sp is not None : <EOL> delta = sp - old_sp <EOL> func_addr = entry_wrapper . current_function_address <EOL> if self . kb . functions . function ( func_addr ) is None : <EOL> l . error ( "<STR_LIT>" + <EOL> "<STR_LIT>" , <EOL> func_addr ) <EOL> self . kb . functions . function ( func_addr , create = True ) <EOL> self . kb . functions . function ( func_addr , create = True ) . sp_delta = delta <EOL> elif jumpkind == '<STR_LIT>' : <EOL> new_call_stack = entry_wrapper . call_stack <EOL> else : <EOL> if self . project . is_hooked ( exit_target ) : <EOL> hooker = self . project . hooked_by ( exit_target ) <EOL> if not hooker is simuvex . procedures . stubs . UserHook . UserHook : <EOL> new_call_stack = entry_wrapper . call_stack_copy ( ) <EOL> new_call_stack . current_function_address = exit_target <EOL> else : <EOL> new_call_stack = entry_wrapper . call_stack <EOL> else : <EOL> new_call_stack = entry_wrapper . call_stack <EOL> return new_call_stack <EOL> def _create_cfgnode ( self , simrun , call_stack_suffix , func_addr ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( simrun , simuvex . procedures . syscalls . handler . handler ) : <EOL> is_syscall = True <EOL> if simrun . syscall is not None : <EOL> syscall = simrun . syscall . __class__ . __name__ <EOL> else : <EOL> syscall = None <EOL> else : <EOL> is_syscall = False <EOL> syscall = None <EOL> if isinstance ( simrun , simuvex . SimProcedure ) : <EOL> simproc_name = simrun . __class__ . __name__ . split ( '<STR_LIT:.>' ) [ - <NUM_LIT:1> ] <EOL> if simproc_name == "<STR_LIT>" : <EOL> simproc_name = simrun . resolves <EOL> no_ret = False <EOL> if ( syscall is None and simrun . NO_RET ) or ( syscall is not None and simrun . syscall . NO_RET ) : <EOL> no_ret = True <EOL> cfg_node = CFGNode ( simrun . addr , <EOL> None , <EOL> self , <EOL> callstack_key = call_stack_suffix , <EOL> input_state = None , <EOL> simprocedure_name = simproc_name , <EOL> syscall_name = syscall , <EOL> no_ret = no_ret , <EOL> is_syscall = is_syscall , <EOL> syscall = syscall , <EOL> function_address = simrun . addr ) <EOL> else : <EOL> cfg_node = CFGNode ( simrun . addr , <EOL> simrun . irsb . size , <EOL> self , <EOL> callstack_key = call_stack_suffix , <EOL> input_state = None , <EOL> is_syscall = is_syscall , <EOL> syscall = syscall , <EOL> simrun = simrun , <EOL> function_address = func_addr ) <EOL> return cfg_node <EOL> def _detect_loop ( self , sim_run , new_tpl , simrun_key , new_call_stack_suffix , new_addr , new_jumpkind , <EOL> current_exit_wrapper , current_function_addr ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( sim_run , simuvex . SimIRSB ) <EOL> if new_jumpkind . startswith ( "<STR_LIT>" ) : <EOL> return <EOL> if new_tpl == simrun_key : <EOL> l . debug ( "<STR_LIT>" , sim_run ) <EOL> if ( sim_run . addr , sim_run . addr ) not in self . _loop_back_edges_set : <EOL> self . _loop_back_edges_set . add ( ( sim_run . addr , sim_run . addr ) ) <EOL> self . _loop_back_edges . append ( ( simrun_key , new_tpl ) ) <EOL> elif new_jumpkind != "<STR_LIT>" and new_jumpkind != "<STR_LIT>" and current_exit_wrapper . bbl_in_stack ( <EOL> new_call_stack_suffix , current_function_addr , new_addr ) : <EOL> '''<STR_LIT>''' <EOL> next_irsb = self . _nodes [ new_tpl ] <EOL> other_preds = set ( ) <EOL> for node_key , node in self . _nodes_by_addr [ next_irsb . addr ] : <EOL> predecessors = self . graph . predecessors ( node ) <EOL> for pred in predecessors : <EOL> if pred . addr != sim_run . addr : <EOL> other_preds . add ( pred ) <EOL> if len ( other_preds ) > <NUM_LIT:0> : <EOL> is_overlapping = False <EOL> for p in other_preds : <EOL> if isinstance ( p , simuvex . SimIRSB ) : <EOL> if p . addr + p . irsb . size ( ) == sim_run . addr + sim_run . irsb . size ( ) : <EOL> is_overlapping = True <EOL> break <EOL> if is_overlapping : <EOL> self . _overlapped_loop_headers . append ( sim_run ) <EOL> l . debug ( "<STR_LIT>" , sim_run ) <EOL> else : <EOL> if ( sim_run . addr , next_irsb . addr ) not in self . _loop_back_edges_set : <EOL> self . _loop_back_edges_set . add ( ( sim_run . addr , next_irsb . addr ) ) <EOL> self . _loop_back_edges . append ( ( simrun_key , new_tpl ) ) <EOL> l . debug ( "<STR_LIT>" , sim_run , next_irsb ) <EOL> else : <EOL> if ( sim_run . addr , next_irsb . addr ) not in self . _loop_back_edges_set : <EOL> self . _loop_back_edges_set . add ( ( sim_run . addr , next_irsb . addr ) ) <EOL> self . _loop_back_edges . append ( ( simrun_key , new_tpl ) ) <EOL> l . debug ( "<STR_LIT>" , sim_run , next_irsb ) <EOL> def _normalize_loop_backedges ( self ) : <EOL> """<STR_LIT>""" <EOL> loop_backedges = [ ] <EOL> for src_key , dst_key in self . _loop_back_edges : <EOL> src = self . _nodes [ src_key ] <EOL> dst = self . _nodes [ dst_key ] <EOL> loop_backedges . append ( ( src , dst ) ) <EOL> self . _loop_back_edges = loop_backedges <EOL> def _analyze_calling_conventions ( self ) : <EOL> """<STR_LIT>""" <EOL> l . debug ( "<STR_LIT>" ) <EOL> for func in self . kb . functions . values ( ) : <EOL> if func . call_convention is not None : <EOL> continue <EOL> callsites = self . _get_callsites ( func . addr ) <EOL> self . _refine_function_arguments ( func , callsites ) <EOL> func . call_convention = func . _match_cc ( ) <EOL> def _analyze_function_features ( self ) : <EOL> """<STR_LIT>""" <EOL> changes = { <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : [ ] <EOL> } <EOL> for func in self . kb . functions . values ( ) : <EOL> if func . returning is not None : <EOL> continue <EOL> if func . endpoints : <EOL> changes [ '<STR_LIT>' ] . append ( func ) <EOL> func . returning = True <EOL> continue <EOL> if self . project . is_hooked ( func . addr ) : <EOL> hooker = self . project . hooked_by ( func . addr ) <EOL> if hasattr ( hooker , '<STR_LIT>' ) and hooker . NO_RET : <EOL> func . returning = False <EOL> changes [ '<STR_LIT>' ] . append ( func ) <EOL> continue <EOL> tmp_graph = networkx . DiGraph ( func . graph ) <EOL> edges_to_remove = [ ] <EOL> for src , dst , data in tmp_graph . edges_iter ( data = True ) : <EOL> if data [ '<STR_LIT:type>' ] == '<STR_LIT>' : <EOL> edges = [ edge for edge in func . transition_graph . edges ( <EOL> nbunch = [ src ] , data = True <EOL> ) if edge [ <NUM_LIT:2> ] [ '<STR_LIT:type>' ] != '<STR_LIT>' <EOL> ] <EOL> if not edges : <EOL> continue <EOL> target_addr = edges [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . addr <EOL> target_func = self . kb . functions . function ( addr = target_addr ) <EOL> if target_func . returning is False : <EOL> edges_to_remove . append ( ( src , dst ) ) <EOL> for src , dst in edges_to_remove : <EOL> tmp_graph . remove_edge ( src , dst ) <EOL> temporary_local_endpoints = [ a for a in tmp_graph . nodes ( ) <EOL> if tmp_graph . out_degree ( a ) == <NUM_LIT:0> ] <EOL> if not temporary_local_endpoints : <EOL> continue <EOL> all_endpoints_returning = [ ] <EOL> temporary_endpoints = [ ] <EOL> for local_endpoint in temporary_local_endpoints : <EOL> if local_endpoint in func . transition_graph . nodes ( ) : <EOL> out_edges = func . transition_graph . out_edges ( [ local_endpoint ] , data = True ) <EOL> if not out_edges : <EOL> temporary_endpoints . append ( local_endpoint ) <EOL> else : <EOL> for src , dst , data in out_edges : <EOL> t = data [ '<STR_LIT:type>' ] <EOL> if t != '<STR_LIT>' : <EOL> temporary_endpoints . append ( dst ) <EOL> for endpoint in temporary_endpoints : <EOL> if endpoint in func . nodes : <EOL> if isinstance ( endpoint , Function ) : <EOL> all_endpoints_returning . append ( endpoint . returning ) <EOL> else : <EOL> successors = [ dst for _ , dst in func . transition_graph . out_edges ( endpoint ) ] <EOL> all_noret = True <EOL> for suc in successors : <EOL> if isinstance ( suc , Function ) : <EOL> if suc . returning is not False : <EOL> all_noret = False <EOL> else : <EOL> n = self . get_any_node ( endpoint . addr , is_syscall = func . is_syscall ) <EOL> if n : <EOL> if n . no_ret is not True : <EOL> all_noret = False <EOL> if all_noret : <EOL> all_endpoints_returning . append ( True ) <EOL> else : <EOL> all_endpoints_returning . append ( None ) <EOL> else : <EOL> call_target = endpoint <EOL> call_target_func = self . kb . functions . function ( call_target ) <EOL> if call_target_func is None : <EOL> all_endpoints_returning . append ( None ) <EOL> continue <EOL> all_endpoints_returning . append ( call_target_func . returning ) <EOL> if all ( [ i is False for i in all_endpoints_returning ] ) : <EOL> func . returning = False <EOL> changes [ '<STR_LIT>' ] . append ( func ) <EOL> return changes <EOL> def _refine_function_arguments ( self , func , callsites ) : <EOL> """<STR_LIT>""" <EOL> for i , c in enumerate ( callsites ) : <EOL> if c . simprocedure_name is not None : <EOL> continue <EOL> l . debug ( "<STR_LIT>" , repr ( func ) , c . addr , i , len ( callsites ) ) <EOL> blocks_ahead = [ c ] <EOL> blocks_after = [ ] <EOL> successors = self . get_successors_and_jumpkind ( c , excluding_fakeret = False ) <EOL> for s , jk in successors : <EOL> if jk == '<STR_LIT>' : <EOL> blocks_after = [ s ] <EOL> break <EOL> regs_overwritten = set ( ) <EOL> stack_overwritten = set ( ) <EOL> regs_read = set ( ) <EOL> regs_written = set ( ) <EOL> try : <EOL> path = self . project . factory . path ( <EOL> self . project . factory . blank_state ( mode = "<STR_LIT>" , addr = blocks_ahead [ <NUM_LIT:0> ] . addr ) ) <EOL> all_successors = path . next_run . successors + path . next_run . unsat_successors <EOL> if len ( all_successors ) == <NUM_LIT:0> : <EOL> continue <EOL> suc = all_successors [ <NUM_LIT:0> ] <EOL> se = suc . se <EOL> actions = suc . log . actions <EOL> sp = se . exactly_int ( suc . regs . sp , default = <NUM_LIT:0> ) + self . project . arch . call_sp_fix <EOL> for ac in actions : <EOL> if ac . type == "<STR_LIT>" and ac . action == "<STR_LIT>" : <EOL> regs_overwritten . add ( ac . offset ) <EOL> elif ac . type == "<STR_LIT>" and ac . action == "<STR_LIT>" : <EOL> addr = se . exactly_int ( ac . addr . ast , default = <NUM_LIT:0> ) <EOL> if ( self . project . arch . call_pushes_ret and addr >= sp + self . project . arch . bits / <NUM_LIT:8> ) or ( not self . project . arch . call_pushes_ret and addr >= sp ) : <EOL> offset = addr - sp <EOL> stack_overwritten . add ( offset ) <EOL> func . prepared_registers . add ( tuple ( regs_overwritten ) ) <EOL> func . prepared_stack_variables . add ( tuple ( stack_overwritten ) ) <EOL> except ( simuvex . SimError , AngrError ) : <EOL> pass <EOL> try : <EOL> if len ( blocks_after ) : <EOL> path = self . project . factory . path ( <EOL> self . project . factory . blank_state ( mode = "<STR_LIT>" , addr = blocks_after [ <NUM_LIT:0> ] . addr ) ) <EOL> all_successors = path . next_run . successors + path . next_run . unsat_successors <EOL> if len ( all_successors ) == <NUM_LIT:0> : <EOL> continue <EOL> suc = all_successors [ <NUM_LIT:0> ] <EOL> actions = suc . log . actions <EOL> for ac in actions : <EOL> if ac . type == "<STR_LIT>" and ac . action == "<STR_LIT>" and ac . offset not in regs_written : <EOL> regs_read . add ( ac . offset ) <EOL> elif ac . type == "<STR_LIT>" and ac . action == "<STR_LIT>" : <EOL> regs_written . add ( ac . offset ) <EOL> regs_read = self . project . arch . argument_registers . intersection ( regs_read ) <EOL> func . registers_read_afterwards . add ( tuple ( regs_read ) ) <EOL> except ( simuvex . SimError , AngrError ) : <EOL> pass <EOL> def _immediate_dominators ( self , node , target_graph = None , reverse_graph = False ) : <EOL> if target_graph is None : <EOL> target_graph = self . graph <EOL> if node not in target_graph : <EOL> raise AngrCFGError ( '<STR_LIT>' % node ) <EOL> graph = networkx . DiGraph ( target_graph ) <EOL> if reverse_graph : <EOL> for n in target_graph . nodes ( ) : <EOL> graph . add_node ( n ) <EOL> for src , dst in target_graph . edges ( ) : <EOL> graph . add_edge ( dst , src ) <EOL> idom = { node : node } <EOL> order = list ( networkx . dfs_postorder_nodes ( graph , node ) ) <EOL> dfn = { u : i for i , u in enumerate ( order ) } <EOL> order . pop ( ) <EOL> order . reverse ( ) <EOL> def intersect ( u_ , v_ ) : <EOL> while u_ != v_ : <EOL> while dfn [ u_ ] < dfn [ v_ ] : <EOL> u_ = idom [ u_ ] <EOL> while dfn [ u_ ] > dfn [ v_ ] : <EOL> v_ = idom [ v_ ] <EOL> return u_ <EOL> changed = True <EOL> while changed : <EOL> changed = False <EOL> for u in order : <EOL> new_idom = reduce ( intersect , ( v for v in graph . pred [ u ] if v in idom ) ) <EOL> if u not in idom or idom [ u ] != new_idom : <EOL> idom [ u ] = new_idom <EOL> changed = True <EOL> return idom <EOL> @ staticmethod <EOL> def _is_indirect_jump ( _ , simirsb ) : <EOL> """<STR_LIT>""" <EOL> if simirsb . irsb . direct_next : <EOL> return False <EOL> if not ( simirsb . irsb . jumpkind == '<STR_LIT>' or simirsb . irsb . jumpkind == '<STR_LIT>' ) : <EOL> return False <EOL> return True <EOL> @ staticmethod <EOL> def _generate_simrun_key ( call_stack_suffix , simrun_addr , is_syscall ) : <EOL> if not is_syscall : <EOL> return call_stack_suffix + ( simrun_addr , '<STR_LIT>' ) <EOL> else : <EOL> return call_stack_suffix + ( simrun_addr , '<STR_LIT>' ) <EOL> @ staticmethod <EOL> def _simrun_key_repr ( simrun_key ) : <EOL> runtype = simrun_key [ - <NUM_LIT:1> ] <EOL> addr = simrun_key [ - <NUM_LIT:2> ] <EOL> callstack = [ ] <EOL> for i in xrange ( <NUM_LIT:0> , len ( simrun_key ) - <NUM_LIT:2> , <NUM_LIT:2> ) : <EOL> from_ = '<STR_LIT:None>' if simrun_key [ i ] is None else hex ( simrun_key [ i ] ) <EOL> to_ = '<STR_LIT:None>' if simrun_key [ i + <NUM_LIT:1> ] is None else hex ( simrun_key [ i + <NUM_LIT:1> ] ) <EOL> callstack . append ( "<STR_LIT>" % ( from_ , to_ ) ) <EOL> s = "<STR_LIT>" % ( "<STR_LIT:U+002CU+0020>" . join ( callstack ) , hex ( addr ) , runtype ) <EOL> return s <EOL> @ staticmethod <EOL> def _simrun_key_callstack_key ( simrun_key ) : <EOL> return simrun_key [ : - <NUM_LIT:2> ] <EOL> @ staticmethod <EOL> def _simrun_key_addr ( simrun_key ) : <EOL> return simrun_key [ - <NUM_LIT:2> ] <EOL> @ staticmethod <EOL> def _simrun_key_current_func_addr ( simrun_key ) : <EOL> """<STR_LIT>""" <EOL> if len ( simrun_key ) > <NUM_LIT:2> : <EOL> return simrun_key [ - <NUM_LIT:3> ] <EOL> else : <EOL> return None <EOL> def _push_unresolvable_run ( self , simrun_address ) : <EOL> self . _unresolvable_runs . add ( simrun_address ) <EOL> def _is_address_executable ( self , address ) : <EOL> """<STR_LIT>""" <EOL> for r in self . _executable_address_ranges : <EOL> if address >= r [ <NUM_LIT:0> ] and address < r [ <NUM_LIT:1> ] : <EOL> return True <EOL> return False <EOL> def _update_thumb_addrs ( self , simrun , state ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( simrun , simuvex . SimIRSB ) and state . thumb : <EOL> self . _thumb_addrs . update ( simrun . imark_addrs ( ) ) <EOL> self . _thumb_addrs . update ( map ( lambda x : x + <NUM_LIT:1> , simrun . imark_addrs ( ) ) ) <EOL> def _get_callsites ( self , function_address ) : <EOL> """<STR_LIT>""" <EOL> all_predecessors = [ ] <EOL> nodes = self . get_all_nodes ( function_address ) <EOL> for n in nodes : <EOL> predecessors = self . get_predecessors ( n ) <EOL> all_predecessors . extend ( predecessors ) <EOL> return all_predecessors <EOL> def _get_nx_paths ( self , begin , end ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( begin , int ) and isinstance ( end , int ) : <EOL> n_begin = self . get_any_node ( begin ) <EOL> n_end = self . get_any_node ( end ) <EOL> elif isinstance ( begin , CFGNode ) and isinstance ( end , CFGNode ) : <EOL> n_begin = begin <EOL> n_end = end <EOL> else : <EOL> raise AngrCFGError ( "<STR_LIT>" ) <EOL> self . remove_fakerets ( ) <EOL> return networkx . all_shortest_paths ( self . graph , n_begin , n_end ) <EOL> def _quasi_topological_sort ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _quasi_topological_order = { } <EOL> ctr = self . _graph . number_of_nodes ( ) <EOL> for ep in self . _entry_points : <EOL> ep_node = self . get_any_node ( ep ) <EOL> if not ep_node : <EOL> continue <EOL> for n in networkx . dfs_postorder_nodes ( self . _graph , source = ep_node ) : <EOL> if n not in self . _quasi_topological_order : <EOL> self . _quasi_topological_order [ n ] = ctr <EOL> ctr -= <NUM_LIT:1> <EOL> CFG = CFGAccurate <EOL> register_analysis ( CFGAccurate , '<STR_LIT>' ) <EOL> register_analysis ( CFGAccurate , '<STR_LIT>' ) </s>
