<commit_before>#!/usr/bin/env python
'''This module handles constraint generation.'''

import symexec
import pyvex

# importing stuff into the module namespace
from s_value import ConcretizingException
from s_irsb import SimIRSB, SimIRSBError
from s_irstmt import SimIRStmt
from s_exit import SimExit
from s_state import SimState
from s_memory import SimMemory, SimMemoryError
from s_exception import SimError
from s_value import SimValue
from s_slice import SimSlice
from s_ref import *
from s_procedure import SimProcedure
import procedures
from s_arch import *
import s_options as o

import logging
l = logging.getLogger("simuvex")

def handle_exit_concrete(base, concrete_start, current_exit, bytes):
	byte_start = concrete_start - base
	irsb = pyvex.IRSB(bytes = bytes[byte_start:], mem_addr = base + byte_start, arch=current_exit.state.arch.vex_arch, basic=True)
	sirsb = SimIRSB(irsb=irsb, initial_state=current_exit.state)
	return sirsb

def concretize_exit(current_exit, fallback_state):
	sat_level = "constrained"

	# get the concrete value
	# TODO: deal with possibility of multiple exits
	l.debug("Concretizing start value...")

	#exit_state = current_exit.state

	# TODO: partial constraining
	#if not current_exit.reachable():
	#	l.warning("UNSAT exit condition. Falling back to fallback state.")
	#	sat_level = "fallback"
	#	current_exit.state = fallback_state

	#if not current_exit.reachable():
	#	l.warning("UNSAT exit condition. Falling back to unconstrained state.")
	#	sat_level = "unconstrained"
	#	current_exit.state = exit_state.copy_after()
	#	current_exit.state.clear_constraints()

	if not current_exit.reachable():
		l.warning("UNSAT exit condition with fallback state. Aborting.")
		return "unsat", None

	# now figure out how many values the exit has
	try:
		concrete_starts = [ current_exit.concretize() ]
	except ConcretizingException:
		max_multiple = 256
		concrete_starts = current_exit.concretize_n(max_multiple)

		l.debug("Got %d possibilities for exit." % len(concrete_starts))
		if len(concrete_starts) == max_multiple:
			l.warning("Exit concretized into the maximum number of targets. Ignoring.")
			concrete_starts = [ ]

	return sat_level, concrete_starts

def handle_exit(base, bytes, current_exit, fallback_state, visited_paths):
	exits_out = [ ]
	exits = [ ]
	sirsbs = { }
	sat_level, concrete_starts = concretize_exit(current_exit, fallback_state)

	if sat_level == "unsat" or len(concrete_starts) == 0:
		l.warning("Got no concrete values for the exit.")
	else:
		l.debug("Got SAT level: %s" % sat_level)
		for concrete_start in concrete_starts:
			l.debug("... concretized start: 0x%x" % concrete_start)
			byte_start = concrete_start - base
			if byte_start < 0 or byte_start >= len(bytes):
				l.info("Exit to 0x%x, outside of the provided bytes." % concrete_start)
				exits_out.append(current_exit)
				continue

			# TODO: more intelligent condition here to handle loops
			if concrete_start not in visited_paths:
				l.debug("... processing block")
				visited_paths.add(concrete_start)
				# Here it might raise exception inside pysex if we encounter
				# some instructions that VEX doesn't understand.
				# Let's catch it here to minimize its influences, so the
				# whole function that we have analyzed up to now will still be preserved.
				try:
					sirsb = handle_exit_concrete(base, concrete_start, current_exit, bytes)
				except SimIRSBError:
					l.warning("Sim IRSB error caught. Skipping this one.", exc_info=True)
					continue

				sirsbs[concrete_start] = sirsb
				new_exits = sirsb.exits()
				l.debug("Got %d exits" % len(new_exits))
				exits.extend(new_exits)

	return sat_level, sirsbs, exits, exits_out



def translate_bytes(base, bytes, entry, initial_state = None, arch="AMD64"):
	l.debug("Translating %d bytes, starting from 0x%x" % (len(bytes), entry))
	remaining_exits = { }
	blocks = { }
	exit_types = ("constrained", "fallback", "unconstrained", "unsat")

	visited_paths = set()
	unsat_exits = [ ]
	exits_out = [ ]

	# take an initial exit
	if initial_state:
		l.debug("Received initial state.")

	entry_state = initial_state if initial_state else SimState(arch=arch)
	entry_point = SimExit(addr = entry, addr_state = entry_state)

	for exit_type in exit_types:
		remaining_exits[exit_type] = [ ]
		blocks[exit_type] = { }

	# TODO: maybe it's more appropriate to track how constrained it's been until now
	remaining_exits["constrained"].append(entry_point)

	# and go!
	for exit_type in exit_types:
		while remaining_exits[exit_type]:
			current_exit = remaining_exits[exit_type].pop()
			sat_level, irsbs, new_exits, new_exits_out = handle_exit(base, bytes, current_exit, entry_state.copy_after(), visited_paths)

			if sat_level == "unsat":
				l.debug("Appending %s to unsat." % current_exit)
				unsat_exits.append(current_exit)

			# an exit can't re-constrain itself it it was deemed unconstrained earlier
			if exit_type != "constrained" and sat_level == "constrained":
				sat_level = exit_type

			blocks[sat_level].update(irsbs)
			remaining_exits[sat_level].extend(new_exits)
			exits_out.extend(new_exits_out)

	return blocks, exits_out, unsat_exits<commit_msg>oops, forgot to rename argument name<commit_after>#!/usr/bin/env python
'''This module handles constraint generation.'''

import symexec
import pyvex

# importing stuff into the module namespace
from s_value import ConcretizingException
from s_irsb import SimIRSB, SimIRSBError
from s_irstmt import SimIRStmt
from s_exit import SimExit
from s_state import SimState
from s_memory import SimMemory, SimMemoryError
from s_exception import SimError
from s_value import SimValue
from s_slice import SimSlice
from s_ref import *
from s_procedure import SimProcedure
import procedures
from s_arch import *
import s_options as o

import logging
l = logging.getLogger("simuvex")

def handle_exit_concrete(base, concrete_start, current_exit, bytes):
	byte_start = concrete_start - base
	irsb = pyvex.IRSB(bytes = bytes[byte_start:], mem_addr = base + byte_start, arch=current_exit.state.arch.vex_arch, basic=True)
	sirsb = SimIRSB(irsb=irsb, initial_state=current_exit.state)
	return sirsb

def concretize_exit(current_exit, fallback_state):
	sat_level = "constrained"

	# get the concrete value
	# TODO: deal with possibility of multiple exits
	l.debug("Concretizing start value...")

	#exit_state = current_exit.state

	# TODO: partial constraining
	#if not current_exit.reachable():
	#	l.warning("UNSAT exit condition. Falling back to fallback state.")
	#	sat_level = "fallback"
	#	current_exit.state = fallback_state

	#if not current_exit.reachable():
	#	l.warning("UNSAT exit condition. Falling back to unconstrained state.")
	#	sat_level = "unconstrained"
	#	current_exit.state = exit_state.copy_after()
	#	current_exit.state.clear_constraints()

	if not current_exit.reachable():
		l.warning("UNSAT exit condition with fallback state. Aborting.")
		return "unsat", None

	# now figure out how many values the exit has
	try:
		concrete_starts = [ current_exit.concretize() ]
	except ConcretizingException:
		max_multiple = 256
		concrete_starts = current_exit.concretize_n(max_multiple)

		l.debug("Got %d possibilities for exit." % len(concrete_starts))
		if len(concrete_starts) == max_multiple:
			l.warning("Exit concretized into the maximum number of targets. Ignoring.")
			concrete_starts = [ ]

	return sat_level, concrete_starts

def handle_exit(base, bytes, current_exit, fallback_state, visited_paths):
	exits_out = [ ]
	exits = [ ]
	sirsbs = { }
	sat_level, concrete_starts = concretize_exit(current_exit, fallback_state)

	if sat_level == "unsat" or len(concrete_starts) == 0:
		l.warning("Got no concrete values for the exit.")
	else:
		l.debug("Got SAT level: %s" % sat_level)
		for concrete_start in concrete_starts:
			l.debug("... concretized start: 0x%x" % concrete_start)
			byte_start = concrete_start - base
			if byte_start < 0 or byte_start >= len(bytes):
				l.info("Exit to 0x%x, outside of the provided bytes." % concrete_start)
				exits_out.append(current_exit)
				continue

			# TODO: more intelligent condition here to handle loops
			if concrete_start not in visited_paths:
				l.debug("... processing block")
				visited_paths.add(concrete_start)
				# Here it might raise exception inside pysex if we encounter
				# some instructions that VEX doesn't understand.
				# Let's catch it here to minimize its influences, so the
				# whole function that we have analyzed up to now will still be preserved.
				try:
					sirsb = handle_exit_concrete(base, concrete_start, current_exit, bytes)
				except SimIRSBError:
					l.warning("Sim IRSB error caught. Skipping this one.", exc_info=True)
					continue

				sirsbs[concrete_start] = sirsb
				new_exits = sirsb.exits()
				l.debug("Got %d exits" % len(new_exits))
				exits.extend(new_exits)

	return sat_level, sirsbs, exits, exits_out



def translate_bytes(base, bytes, entry, initial_state = None, arch="AMD64"):
	l.debug("Translating %d bytes, starting from 0x%x" % (len(bytes), entry))
	remaining_exits = { }
	blocks = { }
	exit_types = ("constrained", "fallback", "unconstrained", "unsat")

	visited_paths = set()
	unsat_exits = [ ]
	exits_out = [ ]

	# take an initial exit
	if initial_state:
		l.debug("Received initial state.")

	entry_state = initial_state if initial_state else SimState(arch=arch)
	entry_point = SimExit(addr = entry, state = entry_state)

	for exit_type in exit_types:
		remaining_exits[exit_type] = [ ]
		blocks[exit_type] = { }

	# TODO: maybe it's more appropriate to track how constrained it's been until now
	remaining_exits["constrained"].append(entry_point)

	# and go!
	for exit_type in exit_types:
		while remaining_exits[exit_type]:
			current_exit = remaining_exits[exit_type].pop()
			sat_level, irsbs, new_exits, new_exits_out = handle_exit(base, bytes, current_exit, entry_state.copy_after(), visited_paths)

			if sat_level == "unsat":
				l.debug("Appending %s to unsat." % current_exit)
				unsat_exits.append(current_exit)

			# an exit can't re-constrain itself it it was deemed unconstrained earlier
			if exit_type != "constrained" and sat_level == "constrained":
				sat_level = exit_type

			blocks[sat_level].update(irsbs)
			remaining_exits[sat_level].extend(new_exits)
			exits_out.extend(new_exits_out)

	return blocks, exits_out, unsat_exits
{'fraction_non_alphanumeric': 0.09406419180201082, 'fraction_numerical': 0.0019334880123743233, 'mean_word_length': 4.7922732362821945, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, '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': '2805866', 'n_tokens_mistral': 3727, 'n_tokens_neox': 3355, 'n_words': 948}