<commit_before>import os
import claripy
from rex.exploit import Exploit
from povsim import CGCPovSimulator
import tempfile
import compilerex
from angr.state_plugins.trace_additions import ChallRespInfo

import logging
l = logging.getLogger("rex.exploit.cgc.cgc_exploit")
l.setLevel("DEBUG")


def _get_byte(var_name):
    ## XXX TODO FIXME DO NOT DO THIS HOLY SHIT WHAT THE FUCK
    ## see https://github.com/angr/angr/issues/922
    idx = var_name.split("_")[2]
    return int(idx, 16)


class CGCFormula(object):
    def __init__(self, solver):
        # the statement
        self.smt_stmt = None
        # map of variable names to boolector ids
        self.name_to_id = dict()
        # map of original variable names to cleaned names
        self.name_mapping = dict()
        # the name of the btor
        self.name = None

        self._clean_formula(solver)

    def _clean_formula(self, solver):
        formula = solver._get_solver().to_smt2()

        # replace occurences of stdin with byte variables
        bytes_seen = []
        for var_name in solver.variables:
            if var_name.startswith("aeg_stdin"):
                byte_name = "byte_%x" % _get_byte(var_name)
                bytes_seen.append(byte_name)
                formula = formula.replace(var_name, byte_name)
                self.name_mapping[var_name] = byte_name

        # remove check-sat, clean up string
        new_form = ""
        for line in formula.split("\n")[2:][:-2]:
            if "declare-fun" not in line:
                new_form += "\"%s\"\n" % (line + "\\n")

        # re-declare the variables, to enforce order
        fmt = "\"(declare-fun %s () (_ BitVec %d))\\n\""
        declarations = []
        for b in bytes_seen:
            declarations.append(fmt % (b, 8))
            self.name_to_id[b] = len(self.name_to_id) + 2

        # gather any other variables
        for line in formula.split("\n")[2:][:-2]:
            if "declare-fun" in line:
                variable = line.split()[1]
                if variable.startswith("byte"):
                    continue
                declarations.append("\"" + line + "\\n\"")
                self.name_mapping[variable] = variable
                self.name_to_id[variable] = len(self.name_to_id) + 2

        declarations = '\n'.join(declarations) + "\n"
        self.smt_stmt = declarations + new_form


class IntInfo(object):
    def __init__(self):
        self.start = None
        self.size = None
        self.base = None
        self.var_name = None


class CGCExploit(Exploit):
    '''
    A CGC exploit object, offers more flexibility than an Exploit object for
    the sake of the game.
    https://github.com/CyberGrandChallenge/cgc-release-documentation/blob/master/walk-throughs/understanding-cfe-povs.md
    '''

    def __init__(self, crash, cgc_type, bypasses_nx, bypasses_aslr):
        super(CGCExploit, self).__init__(crash, bypasses_nx, bypasses_aslr)

        self.cgc_type = cgc_type
        self._raw_payload = None
        self.method_name = 'unclassified'

        # set by all exploits, represents the payload
        self._mem = None

        # arguments CGC infra will send us
        self._arg_vars = None
        # the payload stuff
        self._payload_len = None
        self._raw_payload = None
        # the length of the receiving buffer
        self._recv_buf_len = self.crash.state.solver.eval(self.crash.state.posix.stdout.size)
        # C code which handles challenge response
        self._solver_code = ""
        # the cleaned formulas
        self._formulas = []
        # the integer starts
        self._stdin_int_infos = {}
        self._stdout_int_infos = {}
        self._sorted_stdin_int_infos = []
        self._sorted_stdout_int_infos = []

    # DUMPING

    def dump(self, filename=None):
        '''
        dumps a CGC exploit.

        XXX: just python script for now.
        '''

        return self.dump_python(filename)

    def dump_python(self, filename=None):

        raise NotImplementedError(
            "It is the responsibility of subclasses to implement this method"
            )

    def dump_c(self, filename=None):

        raise NotImplementedError(
            "It is the responsibility of subclasses to implement this method"
            )

    def dump_binary(self, filename=None):
        c_code = self.dump_c()
        compiled_result = compilerex.compile_from_string(c_code)
        if filename is not None:
            with open(filename, 'wb') as f:
                f.write(compiled_result)
            os.chmod(filename, 0o755)
            return None

        return compiled_result

### GENERATION

    def _generate_formula(self, extra_vars_to_solve=None):
        """
        This function is used to generate the equations which are inserted inside C exploits
        """

        if extra_vars_to_solve is None:
            extra_vars_to_solve = []

        st = self.crash.state.copy()

        self._prepare_chall_resp(st)

        # get variables representing stdin
        length = st.solver.eval(st.posix.fd[0].read_pos)

        ft = st.solver._solver._merged_solver_for(lst=[self._mem]+extra_vars_to_solve)
        # filter out constants
        the_vars = set()
        split = ft.split()
        for solver in split:
            if len(solver.variables) > 1:
                the_vars.update(solver.variables)
        ft = st.solver._solver._merged_solver_for(names=the_vars)

        # try to use original crash_input instead of random ones if some input is still symbolic at this stage
        size = st.solver.eval(st.posix.stdin.size)
        for i in range(size):
            byte_value = st.posix.stdin.load(i, 1)
            if not st.solver.symbolic(byte_value):
                continue
            constraint = (byte_value == self.crash.crash[i])
            if not st.solver.satisfiable(extra_constraints=[constraint]):
                continue
            st.solver.add(constraint)

        self._payload_len = length
        self._raw_payload = ChallRespInfo.atoi_dumps(st)

        self._create_solvers(ft, extra_vars_to_solve)

    def _create_solvers(self, ft, extra_vars_to_solve=None):
        split_solvers = ft.split()

        if extra_vars_to_solve is None:
            extra_vars_to_solve = []

        # make sure there is a chall_resp_info plugin
        if not self.crash.state.has_plugin("chall_resp_info"):
            # register a blank one
            self.crash.state.register_plugin("chall_resp_info", ChallRespInfo())
<commit_msg>take in account a crash type that does not a crash input<commit_after>import os
import claripy
from rex.exploit import Exploit
from povsim import CGCPovSimulator
import tempfile
import compilerex
from angr.state_plugins.trace_additions import ChallRespInfo

import logging
l = logging.getLogger("rex.exploit.cgc.cgc_exploit")
l.setLevel("DEBUG")


def _get_byte(var_name):
    ## XXX TODO FIXME DO NOT DO THIS HOLY SHIT WHAT THE FUCK
    ## see https://github.com/angr/angr/issues/922
    idx = var_name.split("_")[2]
    return int(idx, 16)


class CGCFormula(object):
    def __init__(self, solver):
        # the statement
        self.smt_stmt = None
        # map of variable names to boolector ids
        self.name_to_id = dict()
        # map of original variable names to cleaned names
        self.name_mapping = dict()
        # the name of the btor
        self.name = None

        self._clean_formula(solver)

    def _clean_formula(self, solver):
        formula = solver._get_solver().to_smt2()

        # replace occurences of stdin with byte variables
        bytes_seen = []
        for var_name in solver.variables:
            if var_name.startswith("aeg_stdin"):
                byte_name = "byte_%x" % _get_byte(var_name)
                bytes_seen.append(byte_name)
                formula = formula.replace(var_name, byte_name)
                self.name_mapping[var_name] = byte_name

        # remove check-sat, clean up string
        new_form = ""
        for line in formula.split("\n")[2:][:-2]:
            if "declare-fun" not in line:
                new_form += "\"%s\"\n" % (line + "\\n")

        # re-declare the variables, to enforce order
        fmt = "\"(declare-fun %s () (_ BitVec %d))\\n\""
        declarations = []
        for b in bytes_seen:
            declarations.append(fmt % (b, 8))
            self.name_to_id[b] = len(self.name_to_id) + 2

        # gather any other variables
        for line in formula.split("\n")[2:][:-2]:
            if "declare-fun" in line:
                variable = line.split()[1]
                if variable.startswith("byte"):
                    continue
                declarations.append("\"" + line + "\\n\"")
                self.name_mapping[variable] = variable
                self.name_to_id[variable] = len(self.name_to_id) + 2

        declarations = '\n'.join(declarations) + "\n"
        self.smt_stmt = declarations + new_form


class IntInfo(object):
    def __init__(self):
        self.start = None
        self.size = None
        self.base = None
        self.var_name = None


class CGCExploit(Exploit):
    '''
    A CGC exploit object, offers more flexibility than an Exploit object for
    the sake of the game.
    https://github.com/CyberGrandChallenge/cgc-release-documentation/blob/master/walk-throughs/understanding-cfe-povs.md
    '''

    def __init__(self, crash, cgc_type, bypasses_nx, bypasses_aslr):
        super(CGCExploit, self).__init__(crash, bypasses_nx, bypasses_aslr)

        self.cgc_type = cgc_type
        self._raw_payload = None
        self.method_name = 'unclassified'

        # set by all exploits, represents the payload
        self._mem = None

        # arguments CGC infra will send us
        self._arg_vars = None
        # the payload stuff
        self._payload_len = None
        self._raw_payload = None
        # the length of the receiving buffer
        self._recv_buf_len = self.crash.state.solver.eval(self.crash.state.posix.stdout.size)
        # C code which handles challenge response
        self._solver_code = ""
        # the cleaned formulas
        self._formulas = []
        # the integer starts
        self._stdin_int_infos = {}
        self._stdout_int_infos = {}
        self._sorted_stdin_int_infos = []
        self._sorted_stdout_int_infos = []

    # DUMPING

    def dump(self, filename=None):
        '''
        dumps a CGC exploit.

        XXX: just python script for now.
        '''

        return self.dump_python(filename)

    def dump_python(self, filename=None):

        raise NotImplementedError(
            "It is the responsibility of subclasses to implement this method"
            )

    def dump_c(self, filename=None):

        raise NotImplementedError(
            "It is the responsibility of subclasses to implement this method"
            )

    def dump_binary(self, filename=None):
        c_code = self.dump_c()
        compiled_result = compilerex.compile_from_string(c_code)
        if filename is not None:
            with open(filename, 'wb') as f:
                f.write(compiled_result)
            os.chmod(filename, 0o755)
            return None

        return compiled_result

### GENERATION

    def _generate_formula(self, extra_vars_to_solve=None):
        """
        This function is used to generate the equations which are inserted inside C exploits
        """

        if extra_vars_to_solve is None:
            extra_vars_to_solve = []

        st = self.crash.state.copy()

        self._prepare_chall_resp(st)

        # get variables representing stdin
        length = st.solver.eval(st.posix.fd[0].read_pos)

        ft = st.solver._solver._merged_solver_for(lst=[self._mem]+extra_vars_to_solve)
        # filter out constants
        the_vars = set()
        split = ft.split()
        for solver in split:
            if len(solver.variables) > 1:
                the_vars.update(solver.variables)
        ft = st.solver._solver._merged_solver_for(names=the_vars)

        # try to use original crash_input instead of random ones if some input is still symbolic at this stage
        size = st.solver.eval(st.posix.stdin.size)

        # there is a weird crash type called FakeCrash has no crash input
        if hasattr(self.crash, 'crash'):
            for i in range(size):
                byte_value = st.posix.stdin.load(i, 1)
                if not st.solver.symbolic(byte_value):
                    continue
                constraint = (byte_value == self.crash.crash[i])
                if not st.solver.satisfiable(extra_constraints=[constraint]):
                    continue
                st.solver.add(constraint)

        self._payload_len = length
        self._raw_payload = ChallRespInfo.atoi_dumps(st)

        self._create_solvers(ft, extra_vars_to_solve)

    def _create_solvers(self, ft, extra_vars_to_solve=None):
        split_solvers = ft.split()

        if extra_vars_to_solve is None:
            extra_vars_to_solve = []

        # make sure there is a chall_resp_info plugin
        if not self.crash.state.has_plugin("chall_resp_info"):
            # register a blank one
            self.crash.state.register_plugin("chall_resp_info", ChallRespInfo())

{'fraction_non_alphanumeric': 0.09296008450916773, 'fraction_numerical': 0.003320003018184562, 'mean_word_length': 2.1587225929456624, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2338117', 'n_tokens_mistral': 4163, 'n_tokens_neox': 3942, 'n_words': 972}