<commit_before>from collections import defaultdict

import networkx

import logging
import simuvex
import angr
from angr.exit_wrapper import SimExitWrapper

l = logging.getLogger(name="angr.cfg")

class CFG(object):
    def __init__(self):
        self._cfg = None
        self._bbl_dict = None
        self._edge_map = None
        self._loop_back_edges = None
        self._overlapped_loop_headers = None
        self._function_manager = None

    def copy(self):
        new_cfg = CFG()
        new_cfg._cfg = networkx.DiGraph(self._cfg)
        new_cfg._bbl_dict = self._bbl_dict.copy()
        new_cfg._edge_map = self._edge_map.copy()
        new_cfg._loop_back_edges = self._loop_back_edges[::]
        new_cfg._overlapped_loop_headers = self._overlapped_loop_headers[::]
        new_cfg._function_manager = self._function_manager
        return new_cfg

    # Construct the CFG from an angr. binary object
    def construct(self, binary, project, avoid_runs=None):
    	avoid_runs = [ ] if avoid_runs is None else avoid_runs

        # Create the function manager
        self._function_manager = angr.FunctionManager(project, binary)

        # Re-create the DiGraph
        self._cfg = networkx.DiGraph()

        # Traverse all the IRSBs, and put them to a dict
        # It's actually a multi-dict, as each SIRSB might have different states
        # on different call predicates
        self._bbl_dict = {}
        entry_point = binary.entry()
        l.debug("Entry point = 0x%x", entry_point)

        # Crawl the binary, create CFG and fill all the refs inside project!
        loaded_state = project.initial_state(mode="static")
        entry_point_exit = project.exit_to(addr=entry_point,
                                           state=loaded_state.copy(),
                                           jumpkind="Ijk_boring")
        exit_wrapper = SimExitWrapper(entry_point_exit)
        remaining_exits = [exit_wrapper]
        traced_sim_blocks = defaultdict(set)
        traced_sim_blocks[exit_wrapper.call_stack_suffix()].add(
            entry_point_exit.concretize())

        self._loop_back_edges = []
        self._overlapped_loop_headers = []

        # For each call, we are always getting two exits: an Ijk_Call that
        # stands for the real call exit, and an Ijk_Ret that is a simulated exit
        # for the retn address. There are certain cases that the control flow
        # never returns to the next instruction of a callsite due to
        # imprecision of the concrete execution. So we save those simulated
        # exits here to increase our code coverage. Of course the real retn from
        # that call always precedes those "fake" retns.
        # Tuple --> (Initial state, call_stack, bbl_stack)
        fake_func_retn_exits = {}
        # A dict to log edges and the jumpkind between each basic block
        exit_targets = defaultdict(list)
        # A dict to record all blocks that returns to a specific address
        retn_target_sources = defaultdict(list)
        # Iteratively analyze every exit
        checked_blocks = set()
        while len(remaining_exits) > 0:
            current_exit_wrapper = remaining_exits.pop()
            current_exit = current_exit_wrapper.sim_exit()
            call_stack_suffix = current_exit_wrapper.call_stack_suffix()
            addr = current_exit.concretize()
            if addr in checked_blocks:
                continue
            checked_blocks.add(addr)
            initial_state = current_exit.state

            try:
				sim_run = project.sim_run(current_exit)
            except simuvex.s_irsb.SimIRSBError as ex:
                # It's a tragedy that we came across some instructions that VEX
                # does not support. I'll create a terminating stub there
                l.error("SimIRSBError occurred(%s). Creating a PathTerminator.", ex)
                sim_run = \
                    simuvex.procedures.SimProcedures["stubs"]["PathTerminator"](
                        initial_state, addr=addr)
            except angr.errors.AngrError as ex:
                l.error("AngrError %s when creating SimRun at 0x%x",
                        ex, addr)
                # We might be on a wrong branch, and is likely to encounter the
                # "No bytes in memory xxx" exception
                # Just ignore it
                sim_run = None

            if sim_run is not None:<commit_msg>rolling back a change I made whilst being stupid<commit_after>from collections import defaultdict

import networkx

import logging
import simuvex
import angr
from angr.exit_wrapper import SimExitWrapper

l = logging.getLogger(name="angr.cfg")

class CFG(object):
    def __init__(self):
        self._cfg = None
        self._bbl_dict = None
        self._edge_map = None
        self._loop_back_edges = None
        self._overlapped_loop_headers = None
        self._function_manager = None

    def copy(self):
        new_cfg = CFG()
        new_cfg._cfg = networkx.DiGraph(self._cfg)
        new_cfg._bbl_dict = self._bbl_dict.copy()
        new_cfg._edge_map = self._edge_map.copy()
        new_cfg._loop_back_edges = self._loop_back_edges[::]
        new_cfg._overlapped_loop_headers = self._overlapped_loop_headers[::]
        new_cfg._function_manager = self._function_manager
        return new_cfg

    # Construct the CFG from an angr. binary object
    def construct(self, binary, project, avoid_runs=None):
    	avoid_runs = [ ] if avoid_runs is None else avoid_runs

        # Create the function manager
        self._function_manager = angr.FunctionManager(project, binary)

        # Re-create the DiGraph
        self._cfg = networkx.DiGraph()

        # Traverse all the IRSBs, and put them to a dict
        # It's actually a multi-dict, as each SIRSB might have different states
        # on different call predicates
        self._bbl_dict = {}
        entry_point = binary.entry()
        l.debug("Entry point = 0x%x", entry_point)

        # Crawl the binary, create CFG and fill all the refs inside project!
        loaded_state = project.initial_state(mode="static")
        entry_point_exit = project.exit_to(addr=entry_point,
                                           state=loaded_state.copy(),
                                           jumpkind="Ijk_boring")
        exit_wrapper = SimExitWrapper(entry_point_exit)
        remaining_exits = [exit_wrapper]
        traced_sim_blocks = defaultdict(set)
        traced_sim_blocks[exit_wrapper.call_stack_suffix()].add(
            entry_point_exit.concretize())

        self._loop_back_edges = []
        self._overlapped_loop_headers = []

        # For each call, we are always getting two exits: an Ijk_Call that
        # stands for the real call exit, and an Ijk_Ret that is a simulated exit
        # for the retn address. There are certain cases that the control flow
        # never returns to the next instruction of a callsite due to
        # imprecision of the concrete execution. So we save those simulated
        # exits here to increase our code coverage. Of course the real retn from
        # that call always precedes those "fake" retns.
        # Tuple --> (Initial state, call_stack, bbl_stack)
        fake_func_retn_exits = {}
        # A dict to log edges and the jumpkind between each basic block
        exit_targets = defaultdict(list)
        # A dict to record all blocks that returns to a specific address
        retn_target_sources = defaultdict(list)
        # Iteratively analyze every exit
        while len(remaining_exits) > 0:
            current_exit_wrapper = remaining_exits.pop()
            current_exit = current_exit_wrapper.sim_exit()
            call_stack_suffix = current_exit_wrapper.call_stack_suffix()
            addr = current_exit.concretize()
            initial_state = current_exit.state

            try:
				sim_run = project.sim_run(current_exit)
            except simuvex.s_irsb.SimIRSBError as ex:
                # It's a tragedy that we came across some instructions that VEX
                # does not support. I'll create a terminating stub there
                l.error("SimIRSBError occurred(%s). Creating a PathTerminator.", ex)
                sim_run = \
                    simuvex.procedures.SimProcedures["stubs"]["PathTerminator"](
                        initial_state, addr=addr)
            except angr.errors.AngrError as ex:
                l.error("AngrError %s when creating SimRun at 0x%x",
                        ex, addr)
                # We might be on a wrong branch, and is likely to encounter the
                # "No bytes in memory xxx" exception
                # Just ignore it
                sim_run = None

            if sim_run is not None:
{'fraction_non_alphanumeric': 0.06715478778171834, 'fraction_numerical': 0.000686420318041414, 'mean_word_length': 2.208073394495413, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '702907', 'n_tokens_mistral': 2452, 'n_tokens_neox': 2366, 'n_words': 721}