diff options
| -rw-r--r-- | example/samples/dse_crackme.c | 10 | ||||
| -rw-r--r-- | example/symbol_exec/dse_crackme.py | 73 | ||||
| -rw-r--r-- | example/symbol_exec/dse_strategies.py | 129 | ||||
| -rw-r--r-- | miasm2/analysis/dse.py | 83 | ||||
| -rw-r--r-- | miasm2/arch/x86/arch.py | 7 | ||||
| -rw-r--r-- | miasm2/arch/x86/sem.py | 44 | ||||
| -rw-r--r-- | miasm2/expression/expression.py | 182 | ||||
| -rw-r--r-- | test/analysis/dse.py | 111 | ||||
| -rw-r--r-- | test/arch/x86/arch.py | 15 | ||||
| -rw-r--r-- | test/expression/expr_cmp.py | 107 | ||||
| -rwxr-xr-x | test/test_all.py | 14 |
11 files changed, 691 insertions, 84 deletions
diff --git a/example/samples/dse_crackme.c b/example/samples/dse_crackme.c index 5fc0faaf..4621d2be 100644 --- a/example/samples/dse_crackme.c +++ b/example/samples/dse_crackme.c @@ -58,13 +58,13 @@ uint16_t crc16(uint16_t seed, unsigned char *buf, size_t len) return tmp; } -uint16_t test() { +uint16_t test(char* fname) { FILE *file; unsigned char buf[0x100] = {0}; size_t read; uint32_t temp; - file = fopen("test.txt", "r"); + file = fopen(fname, "r"); if (file) { read = fread(buf, sizeof(char), 0x100, file); fclose(file); @@ -94,7 +94,11 @@ uint16_t test() { } int main(int argc, char** argv) { - uint16_t result = test(); + if (argc < 2) { + printf("%s <filename>\n", argv[0]); + return -1; + } + uint16_t result = test(argv[1]); if (result == 0x1337) { printf("OK\n"); } else { diff --git a/example/symbol_exec/dse_crackme.py b/example/symbol_exec/dse_crackme.py index 34c39138..9ac4d6d1 100644 --- a/example/symbol_exec/dse_crackme.py +++ b/example/symbol_exec/dse_crackme.py @@ -10,6 +10,7 @@ import os import subprocess from collections import namedtuple from pdb import pm +from tempfile import NamedTemporaryFile from miasm2.jitter.csts import PAGE_READ, PAGE_WRITE from miasm2.analysis.sandbox import Sandbox_Linux_x86_64 @@ -19,6 +20,8 @@ from miasm2.expression.expression import * my_FILE_ptr = 0x11223344 FInfo = namedtuple("FInfo", ["path", "fdesc"]) FILE_to_info = {} +TEMP_FILE = NamedTemporaryFile() + def xxx_fopen(jitter): ''' #include <stdio.h> @@ -57,8 +60,13 @@ def xxx_fclose(jitter): # Create sandbox parser = Sandbox_Linux_x86_64.parser(description="ELF sandboxer") parser.add_argument("filename", help="ELF Filename") +parser.add_argument("--strategy", + choices=["code-cov", "branch-cov", "path-cov"], + help="Strategy to use for solution creation", + default="code-cov") options = parser.parse_args() options.mimic_env = True +options.command_line = ["%s" % TEMP_FILE.name] sb = Sandbox_Linux_x86_64(options.filename, options, globals()) # Init segment @@ -209,44 +217,17 @@ def xxx_puts_symb(dse): raise FinishOn(string) -done = set([]) # Set of jump address already handled todo = set([""]) # Set of file content to test -class DSEGenFile(DSEPathConstraint): - """DSE with a specific solution creation: - The solution is the content of the FILE to be read - - The politics of exploration is the branch coverage: create a solution only - if the target address has never been seen - """ - - def handle_solution(self, model, destination): - global todo, done - - if destination in done: - # Skip this path, already treated - return - - finfo = FILE_to_info_symb[FILE_stream] - - # Build corresponding file - out = "" - fsize = max(model.eval(self.z3_trans.from_expr(FILE_size)).as_long(), - len(finfo.gen_bytes)) - for index in xrange(fsize): - try: - byteid = finfo.gen_bytes[index] - out += chr(model.eval(self.z3_trans.from_expr(byteid)).as_long()) - except (KeyError, AttributeError) as _: - # Default value if there is no constraint on current byte - out += "\x00" - - todo.add(out) - done.add(destination) - # Instanciate the DSE engine machine = Machine("x86_64") -dse = DSEGenFile(machine) +# Convert strategy to the correct value +strategy = { + "code-cov": DSEPathConstraint.PRODUCE_SOLUTION_CODE_COV, + "branch-cov": DSEPathConstraint.PRODUCE_SOLUTION_BRANCH_COV, + "path-cov": DSEPathConstraint.PRODUCE_SOLUTION_PATH_COV, +}[options.strategy] +dse = DSEPathConstraint(machine, produce_solution=strategy) # Attach to the jitter dse.attach(sb.jitter) @@ -279,8 +260,8 @@ while todo: # Prepare a solution to try, based on the clean state file_content = todo.pop() print "CUR: %r" % file_content - open("test.txt", "w").write(file_content) - dse.restore_snapshot(snapshot) + open(TEMP_FILE.name, "w").write(file_content) + dse.restore_snapshot(snapshot, keep_known_solutions=True) FILE_to_info.clear() FILE_to_info_symb.clear() @@ -294,13 +275,31 @@ while todo: found = True break + finfo = FILE_to_info_symb[FILE_stream] + for sol_ident, model in dse.new_solutions.iteritems(): + # Build the file corresponding to solution in 'model' + + out = "" + fsize = max(model.eval(dse.z3_trans.from_expr(FILE_size)).as_long(), + len(finfo.gen_bytes)) + for index in xrange(fsize): + try: + byteid = finfo.gen_bytes[index] + out += chr(model.eval(dse.z3_trans.from_expr(byteid)).as_long()) + except (KeyError, AttributeError) as _: + # Default value if there is no constraint on current byte + out += "\x00" + + todo.add(out) + # Assert that the result has been found assert found == True print "FOUND !" # Replay for real print "Trying to launch the binary without Miasm" -crackme = subprocess.Popen([options.filename], stdout=subprocess.PIPE, +crackme = subprocess.Popen([options.filename, TEMP_FILE.name], + stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout, stderr = crackme.communicate() assert not stderr diff --git a/example/symbol_exec/dse_strategies.py b/example/symbol_exec/dse_strategies.py new file mode 100644 index 00000000..40998425 --- /dev/null +++ b/example/symbol_exec/dse_strategies.py @@ -0,0 +1,129 @@ +"""Example of DynamicSymbolicExecution engine use + +This example highlights how coverage can be obtained on a binary + +Expected target: 'simple_test.bin' + +Global overview: + - Prepare a 'jitter' instance with the targeted function + - Attach a DSE instance + - Make the function argument symbolic, to track constraints on it + - Take a snapshot + - Initialize the argument candidate list with an arbitrary value, 0 + - Main loop: + - Restore the snapshot (initial state, before running the function) + - Take an argument candidate and set it in the jitter + - Run the function + - Ask the DSE for new candidates, according to its strategy, ie. finding new +block / branch / path +""" +from argparse import ArgumentParser + +from miasm2.analysis.machine import Machine +from miasm2.jitter.csts import PAGE_READ, PAGE_WRITE +from miasm2.analysis.dse import DSEPathConstraint +from miasm2.expression.expression import ExprMem, ExprId, ExprInt, ExprAff + +# Argument handling +parser = ArgumentParser("DSE Example") +parser.add_argument("filename", help="Target x86 shellcode") +parser.add_argument("strategy", choices=["code-cov", "branch-cov", "path-cov"], + help="Strategy to use for solution creation") +args = parser.parse_args() + +# Convert strategy to the correct value +strategy = { + "code-cov": DSEPathConstraint.PRODUCE_SOLUTION_CODE_COV, + "branch-cov": DSEPathConstraint.PRODUCE_SOLUTION_BRANCH_COV, + "path-cov": DSEPathConstraint.PRODUCE_SOLUTION_PATH_COV, +}[args.strategy] + +# Map the shellcode +run_addr = 0x40000 +machine = Machine("x86_32") +jitter = machine.jitter("python") +with open(args.filename) as fdesc: + jitter.vm.add_memory_page(run_addr, PAGE_READ | PAGE_WRITE, fdesc.read(), + "Binary") + +# Expect a binary with one argument on the stack +jitter.init_stack() + +# Argument +jitter.push_uint32_t(0) + +# Handle return +def code_sentinelle(jitter): + jitter.run = False + return False + +ret_addr = 0x1337beef +jitter.add_breakpoint(ret_addr, code_sentinelle) +jitter.push_uint32_t(ret_addr) + +# Init the jitter +jitter.init_run(run_addr) + +# Init a DSE instance with a given strategy +dse = DSEPathConstraint(machine, produce_solution=strategy) +dse.attach(jitter) +# Concretize everything exept the argument +dse.update_state_from_concrete() +regs = jitter.ir_arch.arch.regs +arg = ExprId("ARG", 32) +arg_addr = ExprMem(ExprInt(jitter.cpu.ESP + 4, regs.ESP.size), arg.size) +dse.update_state({ + # @[ESP + 4] = ARG + arg_addr: arg +}) + +# Explore solutions +todo = set([ExprInt(0, arg.size)]) +done = set() +snapshot = dse.take_snapshot() + +# Only needed for the final output +reachs = set() + +while todo: + # Get the next candidate + arg_value = todo.pop() + + # Avoid using twice the same input + if arg_value in done: + continue + done.add(arg_value) + + print "Run with ARG = %s" % arg_value + # Restore state, while keeping already found solutions + dse.restore_snapshot(snapshot, keep_known_solutions=True) + + # Reinit jitter (reset jitter.run, etc.) + jitter.init_run(run_addr) # TODO degage avec new PR? + + # Set the argument value in the jitter context + jitter.eval_expr(ExprAff(arg_addr, arg_value)) + + # Launch + jitter.continue_run() + + # Obtained solutions are in dse.new_solutions: identifier -> solution model + # The identifier depends on the strategy: + # - block address for code coverage + # - last edge for branch coverage + # - execution path for path coverage + + for sol_ident, model in dse.new_solutions.iteritems(): + print "Found a solution to reach: %s" % str(sol_ident) + # Get the argument to use as a Miasm Expr + sol_value = model.eval(dse.z3_trans.from_expr(arg)).as_long() + sol_expr = ExprInt(sol_value, arg.size) + + # Display info and update storages + print "\tARG = %s" % sol_expr + todo.add(sol_expr) + reachs.add(sol_ident) + +print "Found %d input, to reach %d element of coverage" % (len(done), + len(reachs)) + diff --git a/miasm2/analysis/dse.py b/miasm2/analysis/dse.py index 74cc87e9..d97897d8 100644 --- a/miasm2/analysis/dse.py +++ b/miasm2/analysis/dse.py @@ -26,8 +26,9 @@ On branch, if the decision is symbolic, one can also collect "path constraints" and inverse them to produce new inputs potentially reaching new paths. Basically, this is done through DSEPathConstraint. In order to produce a new -solution, one can extend this class, and override 'handle_solution' to -produce a solution which fit its needs. +solution, one can extend this class, and override 'handle_solution' to produce a +solution which fit its needs. It could avoid computing new solution by +overriding 'produce_solution'. If one is only interested in constraints associated to its path, the option "produce_solution" should be set to False, to speed up emulation. @@ -448,7 +449,8 @@ class DSEPathConstraint(DSEEngine): to a new path. In order to produce a new solution, one can extend this class, and override - 'handle_solution' to produce a solution which fit its needs. + 'handle_solution' to produce a solution which fit its needs. It could avoid + computing new solution by overriding 'produce_solution'. If one is only interested in constraints associated to its path, the option "produce_solution" should be set to False, to speed up emulation. @@ -459,7 +461,15 @@ class DSEPathConstraint(DSEEngine): # Maximum memory size to inject in constraints solving MAX_MEMORY_INJECT = 0x10000 - def __init__(self, machine, produce_solution=True, **kwargs): + # Produce solution strategies + PRODUCE_NO_SOLUTION = 0 + PRODUCE_SOLUTION_CODE_COV = 1 + PRODUCE_SOLUTION_BRANCH_COV = 2 + PRODUCE_SOLUTION_PATH_COV = 3 + + def __init__(self, machine, produce_solution=PRODUCE_SOLUTION_CODE_COV, + known_solutions=None, + **kwargs): """Init a DSEPathConstraint @machine: Machine of the targeted architecture instance @produce_solution: (optional) if set, new solutions will be computed""" @@ -469,10 +479,14 @@ class DSEPathConstraint(DSEEngine): assert z3 is not None # Init PathConstraint specifics structures - self.produce_solution = produce_solution self.cur_solver = z3.Solver() - self.new_solutions = {} # destination -> set of model + self.new_solutions = {} # solution identifier -> solution's model + self._known_solutions = set() # set of solution identifiers self.z3_trans = Translator.to_language("z3") + self._produce_solution_strategy = produce_solution + self._history = None + if produce_solution == self.PRODUCE_SOLUTION_PATH_COV: + self._history = [] # List of addresses in the current path def take_snapshot(self, *args, **kwargs): snap = super(DSEPathConstraint, self).take_snapshot(*args, **kwargs) @@ -481,22 +495,69 @@ class DSEPathConstraint(DSEEngine): snap["cur_constraints"] = self.cur_solver.assertions() return snap - def restore_snapshot(self, snapshot, *args, **kwargs): - super(DSEPathConstraint, self).restore_snapshot(snapshot, *args, - **kwargs) + def restore_snapshot(self, snapshot, keep_known_solutions=True, **kwargs): + """Restore a DSEPathConstraint snapshot + @keep_known_solutions: if set, do not forget solutions already found. + -> They will not appear in 'new_solutions' + """ + super(DSEPathConstraint, self).restore_snapshot(snapshot, **kwargs) self.new_solutions.clear() self.new_solutions.update(snapshot["new_solutions"]) self.cur_solver = z3.Solver() self.cur_solver.add(snapshot["cur_constraints"]) + if not keep_known_solutions: + self._known_solutions.clear() + + def _key_for_solution_strategy(self, destination): + """Return the associated identifier for the current solution strategy""" + if self._produce_solution_strategy == self.PRODUCE_NO_SOLUTION: + # Never produce a solution + return None + elif self._produce_solution_strategy == self.PRODUCE_SOLUTION_CODE_COV: + # Decision based on code coverage + # -> produce a solution if the destination has never been seen + key = destination + + elif self._produce_solution_strategy == self.PRODUCE_SOLUTION_BRANCH_COV: + # Decision based on branch coverage + # -> produce a solution if the current branch has never been take + cur_addr = ExprInt(self.jitter.pc, self.ir_arch.IRDst.size) + key = (cur_addr, destination) + + elif self._produce_solution_strategy == self.PRODUCE_SOLUTION_PATH_COV: + # Decision based on path coverage + # -> produce a solution if the current path has never been take + key = tuple(self._history + [destination]) + else: + raise ValueError("Unknown produce solution strategy") + + return key + + def produce_solution(self, destination): + """Called to determine if a solution for @destination should be test for + satisfiability and computed + @destination: Expr instance of the target @destination + """ + key = self._key_for_solution_strategy(destination) + if key is None: + return False + return key not in self._known_solutions def handle_solution(self, model, destination): """Called when a new solution for destination @destination is founded @model: z3 model instance @destination: Expr instance for an addr which is not on the DSE path """ - self.new_solutions.setdefault(destination, set()).add(model) + key = self._key_for_solution_strategy(destination) + assert key is not None + self.new_solutions[key] = model + self._known_solutions.add(key) def handle(self, cur_addr): + # Update history if needed + if self._produce_solution_strategy == self.PRODUCE_SOLUTION_PATH_COV: + self._history.append(cur_addr) + symb_pc = self.eval_expr(self.ir_arch.IRDst) possibilities = possible_values(symb_pc) if len(possibilities) == 1: @@ -547,7 +608,7 @@ class DSEPathConstraint(DSEEngine): # Add path constraint cur_path_constraint = path_constraint - elif self.produce_solution: + elif self.produce_solution(possibility.value): # Looking for a new solution self.cur_solver.push() for cons in path_constraint: diff --git a/miasm2/arch/x86/arch.py b/miasm2/arch/x86/arch.py index 73c9fc0d..6725f5bc 100644 --- a/miasm2/arch/x86/arch.py +++ b/miasm2/arch/x86/arch.py @@ -4254,6 +4254,8 @@ addop("pcmpgtd", [bs8(0x0f), bs8(0x66), no_xmm_pref] + addop("pcmpgtd", [bs8(0x0f), bs8(0x66), pref_66] + rmmod(xmm_reg, rm_arg_xmm)) +addop("pcmpeqq", [bs8(0x0f), bs8(0x38), bs8(0x29), pref_66] + rmmod(xmm_reg, rm_arg_xmm)) +addop("pcmpgtq", [bs8(0x0f), bs8(0x38), bs8(0x37), pref_66] + rmmod(xmm_reg, rm_arg_xmm)) addop("punpckhbw", [bs8(0x0f), bs8(0x68), no_xmm_pref] + rmmod(mm_reg, rm_arg_mm)) @@ -4355,6 +4357,11 @@ addop("shufps", [bs8(0x0f), bs8(0xc6), no_xmm_pref] + addop("shufpd", [bs8(0x0f), bs8(0xc6), pref_66] + rmmod(xmm_reg, rm_arg_xmm) + [u08]) +addop("aesenc", [bs8(0x0f), bs8(0x38), bs8(0xdc), pref_66] + rmmod(xmm_reg, rm_arg_xmm)) +addop("aesdec", [bs8(0x0f), bs8(0x38), bs8(0xde), pref_66] + rmmod(xmm_reg, rm_arg_xmm)) + +addop("aesenclast", [bs8(0x0f), bs8(0x38), bs8(0xdd), pref_66] + rmmod(xmm_reg, rm_arg_xmm)) +addop("aesdeclast", [bs8(0x0f), bs8(0x38), bs8(0xdf), pref_66] + rmmod(xmm_reg, rm_arg_xmm)) mn_x86.bintree = factor_one_bit(mn_x86.bintree) # mn_x86.bintree = factor_fields_all(mn_x86.bintree) diff --git a/miasm2/arch/x86/sem.py b/miasm2/arch/x86/sem.py index 1fceab30..5d564fb1 100644 --- a/miasm2/arch/x86/sem.py +++ b/miasm2/arch/x86/sem.py @@ -3819,25 +3819,53 @@ def pminud(ir, instr, dst, src): def pcmpeq(_, instr, dst, src, size): e = [] for i in xrange(0, dst.size, size): - test = dst[i:i + size] - src[i:i + size] + test = m2_expr.expr_is_equal(dst[i:i + size], src[i:i + size]) e.append(m2_expr.ExprAff(dst[i:i + size], m2_expr.ExprCond(test, - m2_expr.ExprInt(0, size), - m2_expr.ExprInt(-1, size)))) + m2_expr.ExprInt(-1, size), + m2_expr.ExprInt(0, size)))) + return e, [] + + +def pcmpgt(_, instr, dst, src, size): + e = [] + for i in xrange(0, dst.size, size): + test = m2_expr.expr_is_signed_greater(dst[i:i + size], src[i:i + size]) + e.append(m2_expr.ExprAff(dst[i:i + size], + m2_expr.ExprCond(test, + m2_expr.ExprInt(-1, size), + m2_expr.ExprInt(0, size)))) return e, [] def pcmpeqb(ir, instr, dst, src): return pcmpeq(ir, instr, dst, src, 8) - def pcmpeqw(ir, instr, dst, src): return pcmpeq(ir, instr, dst, src, 16) - def pcmpeqd(ir, instr, dst, src): return pcmpeq(ir, instr, dst, src, 32) +def pcmpeqq(ir, instr, dst, src): + return pcmpeq(ir, instr, dst, src, 64) + + + + +def pcmpgtb(ir, instr, dst, src): + return pcmpgt(ir, instr, dst, src, 8) + +def pcmpgtw(ir, instr, dst, src): + return pcmpgt(ir, instr, dst, src, 16) + +def pcmpgtd(ir, instr, dst, src): + return pcmpgt(ir, instr, dst, src, 32) + +def pcmpgtq(ir, instr, dst, src): + return pcmpgt(ir, instr, dst, src, 64) + + def punpck(_, instr, dst, src, size, off): e = [] @@ -4504,6 +4532,12 @@ mnemo_func = {'mov': mov, "pcmpeqb": pcmpeqb, "pcmpeqw": pcmpeqw, "pcmpeqd": pcmpeqd, + "pcmpeqq": pcmpeqq, + + "pcmpgtb": pcmpgtb, + "pcmpgtw": pcmpgtw, + "pcmpgtd": pcmpgtd, + "pcmpgtq": pcmpgtq, "punpckhbw": punpckhbw, "punpckhwd": punpckhwd, diff --git a/miasm2/expression/expression.py b/miasm2/expression/expression.py index b7b90470..591dc024 100644 --- a/miasm2/expression/expression.py +++ b/miasm2/expression/expression.py @@ -1538,3 +1538,185 @@ def get_expr_mem(expr): ops = set() expr.visit(lambda x: visit_getmem(x, ops)) return ops + + +def _expr_compute_cf(op1, op2): + """ + Get carry flag of @op1 - @op2 + Ref: x86 cf flag + @op1: Expression + @op2: Expression + """ + res = op1 - op2 + cf = (((op1 ^ op2) ^ res) ^ ((op1 ^ res) & (op1 ^ op2))).msb() + return cf + +def _expr_compute_of(op1, op2): + """ + Get overflow flag of @op1 - @op2 + Ref: x86 of flag + @op1: Expression + @op2: Expression + """ + res = op1 - op2 + of = (((op1 ^ res) & (op1 ^ op2))).msb() + return of + +def _expr_compute_zf(op1, op2): + """ + Get zero flag of @op1 - @op2 + @op1: Expression + @op2: Expression + """ + res = op1 - op2 + zf = ExprCond(res, + ExprInt(0, 1), + ExprInt(1, 1)) + return zf + + +def _expr_compute_nf(op1, op2): + """ + Get negative (or sign) flag of @op1 - @op2 + @op1: Expression + @op2: Expression + """ + res = op1 - op2 + nf = res.msb() + return nf + + +def expr_is_equal(op1, op2): + """ + if op1 == op2: + Return ExprInt(1, 1) + else: + Return ExprInt(0, 1) + """ + + zf = _expr_compute_zf(op1, op2) + return zf + + +def expr_is_not_equal(op1, op2): + """ + if op1 != op2: + Return ExprInt(1, 1) + else: + Return ExprInt(0, 1) + """ + + zf = _expr_compute_zf(op1, op2) + return ~zf + + +def expr_is_unsigned_greater(op1, op2): + """ + UNSIGNED cmp + if op1 > op2: + Return ExprInt(1, 1) + else: + Return ExprInt(0, 1) + """ + + cf = _expr_compute_cf(op1, op2) + zf = _expr_compute_zf(op1, op2) + return ~(cf | zf) + + +def expr_is_unsigned_greater_or_equal(op1, op2): + """ + Unsigned cmp + if op1 >= op2: + Return ExprInt(1, 1) + else: + Return ExprInt(0, 1) + """ + + cf = _expr_compute_cf(op1, op2) + return ~cf + + +def expr_is_unsigned_lower(op1, op2): + """ + Unsigned cmp + if op1 < op2: + Return ExprInt(1, 1) + else: + Return ExprInt(0, 1) + """ + + cf = _expr_compute_cf(op1, op2) + return cf + + +def expr_is_unsigned_lower_or_equal(op1, op2): + """ + Unsigned cmp + if op1 <= op2: + Return ExprInt(1, 1) + else: + Return ExprInt(0, 1) + """ + + cf = _expr_compute_cf(op1, op2) + zf = _expr_compute_zf(op1, op2) + return cf | zf + + +def expr_is_signed_greater(op1, op2): + """ + Signed cmp + if op1 > op2: + Return ExprInt(1, 1) + else: + Return ExprInt(0, 1) + """ + + nf = _expr_compute_nf(op1, op2) + of = _expr_compute_of(op1, op2) + zf = _expr_compute_zf(op1, op2) + return ~(zf | (nf ^ of)) + + +def expr_is_signed_greater_or_equal(op1, op2): + """ + Signed cmp + if op1 > op2: + Return ExprInt(1, 1) + else: + Return ExprInt(0, 1) + """ + + nf = _expr_compute_nf(op1, op2) + of = _expr_compute_of(op1, op2) + return ~(nf ^ of) + + +def expr_is_signed_lower(op1, op2): + """ + Signed cmp + if op1 < op2: + Return ExprInt(1, 1) + else: + Return ExprInt(0, 1) + """ + + nf = _expr_compute_nf(op1, op2) + of = _expr_compute_of(op1, op2) + return nf ^ of + + +def expr_is_signed_lower_or_equal(op1, op2): + """ + Signed cmp + if op1 <= op2: + Return ExprInt(1, 1) + else: + Return ExprInt(0, 1) + """ + + nf = _expr_compute_nf(op1, op2) + of = _expr_compute_of(op1, op2) + zf = _expr_compute_zf(op1, op2) + return zf | (nf ^ of) diff --git a/test/analysis/dse.py b/test/analysis/dse.py index 59fc2b59..5a72db34 100644 --- a/test/analysis/dse.py +++ b/test/analysis/dse.py @@ -1,20 +1,17 @@ import sys -import os from pdb import pm -from miasm2.core.cpu import ParseAst -from miasm2.arch.x86.arch import mn_x86, base_expr, variable -from miasm2.core import parse_asm -from miasm2.expression.expression import * -from miasm2.core import asmblock from elfesteem.strpatchwork import StrPatchwork +from miasm2.core import parse_asm +from miasm2.expression.expression import ExprCompose, ExprOp, ExprInt, ExprId +from miasm2.core.asmblock import asm_resolve_final from miasm2.analysis.machine import Machine -from miasm2.jitter.csts import * +from miasm2.jitter.csts import PAGE_READ, PAGE_WRITE from miasm2.analysis.dse import DSEEngine -reg_and_id = dict(mn_x86.regs.all_regs_ids_byname) -class DSE_test(object): +class DSETest(object): + """Inspired from TEST/ARCH/X86 Test the symbolic execution correctly follow generated labels @@ -33,16 +30,22 @@ class DSE_test(object): def __init__(self, jitter_engine): self.machine = Machine(self.arch_name) - self.myjit = self.machine.jitter(jitter_engine) + jitter = self.machine.jitter + self.myjit = jitter(jitter_engine) self.myjit.init_stack() - self.myjit.jit.log_regs = False - self.myjit.jit.log_mn = False + self.myjit.jit.log_regs = True + self.myjit.jit.log_mn = True + + self.dse = None + self.assembly = None def init_machine(self): - self.myjit.vm.add_memory_page(self.run_addr, PAGE_READ | PAGE_WRITE, self.assembly) + self.myjit.vm.add_memory_page(self.run_addr, + PAGE_READ | PAGE_WRITE, + self.assembly) self.myjit.push_uint32_t(self.ret_addr) - self.myjit.add_breakpoint(self.ret_addr, lambda x:False) + self.myjit.add_breakpoint(self.ret_addr, lambda x: False) def prepare(self): self.myjit.cpu.ECX = 4 @@ -63,22 +66,25 @@ class DSE_test(object): self.myjit.init_run(self.run_addr) self.myjit.continue_run() - assert(self.myjit.pc == self.ret_addr) + assert self.myjit.pc == self.ret_addr def asm(self): - blocks, symbol_pool = parse_asm.parse_txt(mn_x86, self.arch_attrib, self.TXT, - symbol_pool=self.myjit.ir_arch.symbol_pool) - + mn_x86 = self.machine.mn + blocks, symbol_pool = parse_asm.parse_txt( + mn_x86, + self.arch_attrib, + self.TXT, + symbol_pool=self.myjit.ir_arch.symbol_pool + ) # fix shellcode addr symbol_pool.set_offset(symbol_pool.getby_name("main"), 0x0) - s = StrPatchwork() - patches = asmblock.asm_resolve_final(mn_x86, blocks, symbol_pool) + output = StrPatchwork() + patches = asm_resolve_final(mn_x86, blocks, symbol_pool) for offset, raw in patches.items(): - s[offset] = raw + output[offset] = raw - s = str(s) - self.assembly = s + self.assembly = str(output) def check(self): regs = self.dse.ir_arch.arch.regs @@ -90,5 +96,62 @@ class DSE_test(object): ExprInt(0x0, 24)) & ExprInt(0x1F, 32)) assert value == expected + +class DSEAttachInBreakpoint(DSETest): + + """ + Test that DSE is "attachable" in a jitter breakpoint + """ + TXT = ''' + main: + MOV EAX, 5 + ADD EBX, 6 + INC EBX + RET + ''' + + def __init__(self, *args, **kwargs): + super(DSEAttachInBreakpoint, self).__init__(*args, **kwargs) + self._dse = None + ircls = self.machine.ir + self._regs = ircls().arch.regs + self._testid = ExprId("TEST", self._regs.EBX.size) + + def bp_attach(self, jitter): + """Attach a DSE in the current jitter""" + self.dse = DSEEngine(self.machine) + self.dse.attach(self.myjit) + self.dse.update_state_from_concrete() + self.dse.update_state({ + self._regs.EBX: self._testid, + }) + + # Additionnal call to the exec callback is necessary, as breakpoints are + # honored AFTER exec callback + jitter.exec_cb(jitter) + + return True + + def prepare(self): + pass + + def init_machine(self): + super(DSEAttachInBreakpoint, self).init_machine() + self.myjit.add_breakpoint(5, self.bp_attach) # On ADD EBX, 6 + + def check(self): + value = self.dse.eval_expr(self._regs.EBX) + # EBX = TEST + # ADD EBX, 6 + # INC EBX + # -> EBX_final = TEST + 7 + assert value == self._testid + ExprInt(7, self._regs.EBX.size) + + if __name__ == "__main__": - [test(*sys.argv[1:])() for test in [DSE_test]] + jit_engine = sys.argv[1] + for test in [ + DSETest, + DSEAttachInBreakpoint, + ]: + test(jit_engine)() diff --git a/test/arch/x86/arch.py b/test/arch/x86/arch.py index 0472e714..76e5a5eb 100644 --- a/test/arch/x86/arch.py +++ b/test/arch/x86/arch.py @@ -2793,6 +2793,12 @@ reg_tests = [ (m64, "00000000 PCMPGTD XMM3, XMM0", "660f66d8"), + (m32, "00000000 PCMPEQQ XMM0, XMM5", + "660f3829C5"), + (m32, "00000000 PCMPGTQ XMM0, XMM5", + "660f3837C5"), + + (m32, "00000000 PUNPCKHBW MM2, QWORD PTR [EDX]", "0F6812"), (m32, "00000000 PUNPCKHBW XMM2, XMMWORD PTR [EDX]", @@ -2939,6 +2945,15 @@ reg_tests = [ (m64, "00000000 SHUFPD XMM0, XMM6, 0x44", "660fc6c644"), + (m32, "00000000 AESENC XMM1, XMM2", + "660f38dcca"), + (m32, "00000000 AESDEC XMM1, XMM2", + "660f38deca"), + + (m32, "00000000 AESENCLAST XMM1, XMM2", + "660f38ddca"), + (m32, "00000000 AESDECLAST XMM1, XMM2", + "660f38dfca"), ] diff --git a/test/expression/expr_cmp.py b/test/expression/expr_cmp.py new file mode 100644 index 00000000..b238151d --- /dev/null +++ b/test/expression/expr_cmp.py @@ -0,0 +1,107 @@ +# +# Expression comparison regression tests # +# +from pdb import pm +from miasm2.expression.expression import ExprInt, expr_is_unsigned_greater,\ + expr_is_unsigned_greater_or_equal, expr_is_unsigned_lower,\ + expr_is_unsigned_lower_or_equal, expr_is_signed_greater,\ + expr_is_signed_greater_or_equal, expr_is_signed_lower, \ + expr_is_signed_lower_or_equal, expr_is_equal, expr_is_not_equal +from miasm2.expression.simplifications import expr_simp + +int_0 = ExprInt(0, 32) +int_1 = ExprInt(1, 32) +int_m1 = ExprInt(-1, 32) +int_m2 = ExprInt(-2, 32) + +b0 = ExprInt(0, 1) +b1 = ExprInt(1, 1) + +tests = [ + # unsigned + (b1, expr_is_unsigned_greater, int_1, int_0), + (b1, expr_is_unsigned_lower, int_0, int_1), + + (b0, expr_is_unsigned_greater, int_0, int_1), + (b0, expr_is_unsigned_lower, int_1, int_0), + + (b1, expr_is_unsigned_greater_or_equal, int_1, int_0), + (b1, expr_is_unsigned_lower_or_equal, int_0, int_1), + + (b0, expr_is_unsigned_greater_or_equal, int_0, int_1), + (b0, expr_is_unsigned_lower_or_equal, int_1, int_0), + + (b1, expr_is_unsigned_greater_or_equal, int_1, int_1), + (b1, expr_is_unsigned_lower_or_equal, int_1, int_1), + + (b1, expr_is_unsigned_greater, int_m1, int_0), + (b1, expr_is_unsigned_lower, int_0, int_m1), + + (b0, expr_is_unsigned_greater, int_0, int_m1), + (b0, expr_is_unsigned_lower, int_m1, int_0), + + + # signed + (b1, expr_is_signed_greater, int_1, int_0), + (b1, expr_is_signed_lower, int_0, int_1), + + (b0, expr_is_signed_greater, int_0, int_1), + (b0, expr_is_signed_lower, int_1, int_0), + + (b1, expr_is_signed_greater_or_equal, int_1, int_0), + (b1, expr_is_signed_lower_or_equal, int_0, int_1), + + (b0, expr_is_signed_greater_or_equal, int_0, int_1), + (b0, expr_is_signed_lower_or_equal, int_1, int_0), + + (b1, expr_is_signed_greater_or_equal, int_1, int_1), + (b1, expr_is_signed_lower_or_equal, int_1, int_1), + + (b0, expr_is_signed_greater, int_m1, int_0), + (b0, expr_is_signed_lower, int_0, int_m1), + + (b1, expr_is_signed_greater, int_0, int_m1), + (b1, expr_is_signed_lower, int_m1, int_0), + + + # greater lesser, neg + (b1, expr_is_signed_greater, int_1, int_m1), + (b1, expr_is_signed_lower, int_m1, int_1), + + (b0, expr_is_signed_greater, int_m1, int_1), + (b0, expr_is_signed_lower, int_1, int_m1), + + (b1, expr_is_signed_greater_or_equal, int_1, int_m1), + (b1, expr_is_signed_lower_or_equal, int_m1, int_1), + + (b0, expr_is_signed_greater_or_equal, int_m1, int_1), + (b0, expr_is_signed_lower_or_equal, int_1, int_m1), + + (b1, expr_is_signed_greater_or_equal, int_m1, int_m1), + (b1, expr_is_signed_lower_or_equal, int_m1, int_m1), + + + (b1, expr_is_signed_greater, int_m1, int_m2), + (b1, expr_is_signed_lower, int_m2, int_m1), + + (b0, expr_is_signed_greater, int_m2, int_m1), + (b0, expr_is_signed_lower, int_m1, int_m2), + + (b1, expr_is_signed_greater_or_equal, int_m1, int_m2), + (b1, expr_is_signed_lower_or_equal, int_m2, int_m1), + + (b0, expr_is_signed_greater_or_equal, int_m2, int_m1), + (b0, expr_is_signed_lower_or_equal, int_m1, int_m2), + + # eq/neq + (b1, expr_is_equal, int_1, int_1), + (b1, expr_is_not_equal, int_0, int_1), + + (b0, expr_is_equal, int_1, int_0), + (b0, expr_is_not_equal, int_0, int_0), + + +] + +for result, func, arg1, arg2 in tests: + assert result == expr_simp(func(arg1, arg2)) diff --git a/test/test_all.py b/test/test_all.py index 17193d9f..d2ae4fce 100755 --- a/test/test_all.py +++ b/test/test_all.py @@ -245,6 +245,7 @@ for script in ["modint.py", "expression_helper.py", "expr_pickle.py", "parser.py", + "expr_cmp.py", ]: testset += RegressionTest([script], base_dir="expression") @@ -630,10 +631,15 @@ dse_crackme = ExampleSymbolExec([Example.get_sample("dse_crackme.c"), products=[dse_crackme_out], executable="cc") testset += dse_crackme -testset += ExampleSymbolExec(["dse_crackme.py", dse_crackme_out], - depends=[dse_crackme], - products=["test.txt"], - tags=[TAGS["z3"]]) +for strategy in ["code-cov", "branch-cov", "path-cov"]: + testset += ExampleSymbolExec(["dse_crackme.py", dse_crackme_out, + "--strategy", strategy], + depends=[dse_crackme], + tags=[TAGS["z3"]]) + testset += ExampleSymbolExec(["dse_strategies.py", + Example.get_sample("simple_test.bin"), + strategy], + tags=[TAGS["z3"]]) ## Jitter class ExampleJitter(Example): |