diff options
Diffstat (limited to 'miasm2/ir/symbexec.py')
| -rw-r--r-- | miasm2/ir/symbexec.py | 1414 |
1 files changed, 1032 insertions, 382 deletions
diff --git a/miasm2/ir/symbexec.py b/miasm2/ir/symbexec.py index 8ecde21c..3cde2af7 100644 --- a/miasm2/ir/symbexec.py +++ b/miasm2/ir/symbexec.py @@ -1,13 +1,12 @@ import warnings import logging +from collections import MutableMapping -import miasm2.expression.expression as m2_expr -from miasm2.expression.modint import int32 +from miasm2.expression.expression import ExprOp, ExprId, ExprInt, ExprMem, \ + ExprCompose, ExprSlice, ExprCond, ExprAff from miasm2.expression.simplifications import expr_simp from miasm2.core import asmblock from miasm2.ir.ir import AssignBlock -from miasm2.core.interval import interval - log = logging.getLogger("symbexec") console_handler = logging.StreamHandler() @@ -28,87 +27,6 @@ def get_block(ir_arch, mdis, addr): return irblock -class SymbolMngr(object): - """ - Store registers and memory symbolic values - """ - - def __init__(self, init=None): - if init is None: - init = {} - self.symbols_id = {} - self.symbols_mem = {} - for expr, value in init.items(): - self[expr] = value - - def __contains__(self, expr): - if not isinstance(expr, m2_expr.ExprMem): - return self.symbols_id.__contains__(expr) - if not self.symbols_mem.__contains__(expr.arg): - return False - return self.symbols_mem[expr.arg][0].size == expr.size - - def __getitem__(self, expr): - if not isinstance(expr, m2_expr.ExprMem): - return self.symbols_id.__getitem__(expr) - if not expr.arg in self.symbols_mem: - raise KeyError(expr) - mem, value = self.symbols_mem.__getitem__(expr.arg) - if mem.size != expr.size: - raise KeyError(expr) - return value - - def get(self, expr, default=None): - if not isinstance(expr, m2_expr.ExprMem): - return self.symbols_id.get(expr, default) - if not expr.arg in self.symbols_mem: - return default - mem, value = self.symbols_mem.__getitem__(expr.arg) - if mem.size != expr.size: - return default - return value - - def __setitem__(self, expr, value): - if not isinstance(expr, m2_expr.ExprMem): - self.symbols_id.__setitem__(expr, value) - return - assert expr.size == value.size - self.symbols_mem.__setitem__(expr.arg, (expr, value)) - - def __iter__(self): - for expr in self.symbols_id: - yield expr - for expr in self.symbols_mem: - yield self.symbols_mem[expr][0] - - def __delitem__(self, expr): - if not isinstance(expr, m2_expr.ExprMem): - self.symbols_id.__delitem__(expr) - else: - self.symbols_mem.__delitem__(expr.arg) - - def items(self): - return self.symbols_id.items() + [x for x in self.symbols_mem.values()] - - def keys(self): - return (self.symbols_id.keys() + - [x[0] for x in self.symbols_mem.values()]) - - def copy(self): - new_symbols = SymbolMngr() - new_symbols.symbols_id = dict(self.symbols_id) - new_symbols.symbols_mem = dict(self.symbols_mem) - return new_symbols - - def inject_info(self, info): - new_symbols = SymbolMngr() - for expr, value in self.items(): - expr = expr_simp(expr.replace_expr(info)) - value = expr_simp(value.replace_expr(info)) - new_symbols[expr] = value - return new_symbols - - class StateEngine(object): """Stores an Engine state""" @@ -140,6 +58,7 @@ class SymbolicState(StateEngine): yield dst, src def iteritems(self): + """Iterate on stored memory/values""" return self.__iter__() def merge(self, other): @@ -163,10 +82,721 @@ class SymbolicState(StateEngine): return dict(self._symbols) +INTERNAL_INTBASE_NAME = "__INTERNAL_INTBASE__" + + +def get_expr_base_offset(expr): + """Return a couple representing the symbolic/concrete part of an addition + expression. + + If there is no symbolic part, ExprId(INTERNAL_INTBASE_NAME) is used + If there is not concrete part, 0 is used + @expr: Expression instance + + """ + if expr.is_int(): + internal_intbase = ExprId(INTERNAL_INTBASE_NAME, expr.size) + return internal_intbase, int(expr) + + if not expr.is_op('+'): + return expr, 0 + if expr.args[-1].is_int(): + args, offset = expr.args[:-1], int(expr.args[-1]) + if len(args) == 1: + return args[0], offset + return ExprOp('+', *args), offset + return expr, 0 + + +class MemArray(MutableMapping): + """Link between base and its (offset, Expr) + + Given an expression (say *base*), this structure will store every memory + content relatively to an integer offset from *base*. + + The value associated to a given offset is a description of the slice of a + stored expression. The slice size depends on the configutation of the + MemArray. For example, for a slice size of 8 bits, the affectation: + - @32[EAX+0x10] = EBX + + will store for the base EAX: + - 0x10: (EBX, 0) + - 0x11: (EBX, 1) + - 0x12: (EBX, 2) + - 0x13: (EBX, 3) + + If the *base* is EAX+EBX, this structure can store the following contents: + - @32[EAX+EBX] + - @8[EAX+EBX+0x100] + But not: + - @32[EAX+0x10] (which is stored in another MemArray based on EAX) + - @32[EAX+EBX+ECX] (which is stored in another MemArray based on + EAX+EBX+ECX) + + """ + + def __init__(self, base, expr_simp=expr_simp): + self._base = base + self.expr_simp = expr_simp + self._mask = int(base.mask) + self._offset_to_expr = {} + + @property + def base(self): + """Expression representing the symbolic base address""" + return self._base + + @property + def mask(self): + """Mask offset""" + return self._mask + + def __getitem__(self, offset): + assert 0 <= offset <= self._mask + return self._offset_to_expr.__getitem__(offset) + + def __setitem__(self, offset, value): + raise RuntimeError("Use write api to update keys") + + def __delitem__(self, offset): + assert 0 <= offset <= self._mask + return self._offset_to_expr.__delitem__(offset) + + def __iter__(self): + for offset, _ in self._offset_to_expr.iteritems(): + yield offset + + def __len__(self): + return len(self._offset_to_expr) + + def __repr__(self): + out = [] + out.append("Base: %s" % self.base) + for offset, (index, value) in sorted(self._offset_to_expr.iteritems()): + out.append("%16X %d %s" % (offset, index, value)) + return '\n'.join(out) + + def copy(self): + """Copy object instance""" + obj = MemArray(self.base, self.expr_simp) + obj._offset_to_expr = self._offset_to_expr.copy() + return obj + + @staticmethod + def offset_to_ptr(base, offset): + """ + Return an expression representing the @base + @offset + @base: symbolic base address + @offset: relative offset integer to the @base address + """ + if base.is_id(INTERNAL_INTBASE_NAME): + ptr = ExprInt(offset, base.size) + elif offset == 0: + ptr = base + else: + ptr = base + ExprInt(offset, base.size) + return ptr + + def read(self, offset, size): + """ + Return memory at @offset with @size as an Expr list + @offset: integer (in bytes) + @size: integer (in bits), byte aligned + + Consider the following state: + - 0x10: (EBX, 0) + - 0x11: (EBX, 1) + - 0x12: (EBX, 2) + - 0x13: (EBX, 3) + + A read at 0x10 of 32 bits should return: EBX + """ + + assert size % 8 == 0 + # Parts is (Expr's offset, size, Expr) + parts = [] + for index in xrange(size / 8): + # Wrap read: + # @32[EAX+0xFFFFFFFF] is ok and will read at 0xFFFFFFFF, 0, 1, 2 + request_offset = (offset + index) & self._mask + if request_offset in self._offset_to_expr: + # Known memory portion + off, data = self._offset_to_expr[request_offset] + parts.append((off, 1, data)) + continue + + # Unknown memory portion + ptr = self.offset_to_ptr(self.base, request_offset) + data = ExprMem(ptr, 8) + parts.append((0, 1, data)) + + # Group similar datas + # XXX TODO: only little endian here + index = 0 + while index + 1 < len(parts): + off_a, size_a, data_a = parts[index] + off_b, size_b, data_b = parts[index+1] + if data_a == data_b and off_a + size_a == off_b: + # Read consecutive bytes of a variable + # [(0, 8, x), (1, 8, x)] => (0, 16, x) + parts[index:index+2] = [(off_a, size_a + size_b, data_a)] + continue + if data_a.is_int() and data_b.is_int(): + # Read integer parts + # [(0, 8, 0x11223344), (1, 8, 0x55667788)] => (0, 16, 0x7744) + int1 = self.expr_simp(data_a[off_a*8:(off_a+size_a)*8]) + int2 = self.expr_simp(data_b[off_b*8:(off_b+size_b)*8]) + assert int1.is_int() and int2.is_int() + int1, int2 = int(int1), int(int2) + result = ExprInt((int2 << (size_a * 8)) | int1, (size_a + size_b) * 8) + parts[index:index+2] = [(0, size_a + size_b, result)] + continue + if data_a.is_mem() and data_b.is_mem(): + # Read consecutive bytes of a memory variable + ptr_base_a, ptr_offset_a = get_expr_base_offset(data_a.arg) + ptr_base_b, ptr_offset_b = get_expr_base_offset(data_b.arg) + if ptr_base_a != ptr_base_b: + index += 1 + continue + if (ptr_offset_a + off_a + size_a) & self._mask == (ptr_offset_b + off_b) & self._mask: + assert size_a <= data_a.size / 8 - off_a + assert size_b <= data_b.size / 8 - off_b + # Successive comparable symbolic pointers + # [(0, 8, @8[ptr]), (0, 8, @8[ptr+1])] => (0, 16, @16[ptr]) + ptr = self.offset_to_ptr(ptr_base_a, (ptr_offset_a + off_a) & self._mask) + + data = ExprMem(ptr, (size_a + size_b) * 8) + parts[index:index+2] = [(0, size_a + size_b, data)] + + continue + + index += 1 + + # Slice datas + read_mem = [] + for off, bytesize, data in parts: + if data.size / 8 != bytesize: + data = data[off * 8: (off + bytesize) * 8] + read_mem.append(data) + + return read_mem + + def write(self, offset, expr): + """ + Write @expr at @offset + @offset: integer (in bytes) + @expr: Expr instance value + """ + assert expr.size % 8 == 0 + assert offset <= self._mask + for index in xrange(expr.size / 8): + # Wrap write: + # @32[EAX+0xFFFFFFFF] is ok and will write at 0xFFFFFFFF, 0, 1, 2 + request_offset = (offset + index) & self._mask + # XXX TODO: only little endian here + self._offset_to_expr[request_offset] = (index, expr) + + tmp = self.expr_simp(expr[index * 8: (index + 1) * 8]) + # Special case: Simplify slice of pointer (simplification is ok + # here, as we won't store the simplified expression) + if tmp.is_slice() and tmp.arg.is_mem() and tmp.start % 8 == 0: + new_ptr = self.expr_simp(tmp.arg.arg + ExprInt(tmp.start / 8, tmp.arg.arg.size)) + tmp = ExprMem(new_ptr, tmp.stop - tmp.start) + # Test if write to original value + if tmp.is_mem(): + src_ptr, src_off = get_expr_base_offset(tmp.arg) + if src_ptr == self.base and src_off == request_offset: + del self._offset_to_expr[request_offset] + + + def _get_variable_parts(self, index, known_offsets, forward=True): + """ + Find consecutive memory parts representing the same variable. The part + starts at offset known_offsets[@index] and search is in offset direction + determined by @forward + Return the number of consecutive parts of the same variable. + + @index: index of the memory offset in known_offsets + @known_offsets: sorted offsets + @forward: Search in offset growing direction if True, else in reverse + order + """ + + offset = known_offsets[index] + value_byte_index, value = self._offset_to_expr[offset] + assert value.size % 8 == 0 + + if forward: + start, end, step = value_byte_index + 1, value.size / 8, 1 + else: + start, end, step = value_byte_index - 1, -1, -1 + + partnum = 1 + for value_offset in xrange(start, end, step): + offset += step + # Check if next part is in known_offsets + next_index = index + step * partnum + if not 0 <= next_index < len(known_offsets): + break + + offset_next = known_offsets[next_index] + if offset_next != offset: + break + + # Check if next part is a part of the searched value + byte_index, value_next = self._offset_to_expr[offset_next] + if byte_index != value_offset: + break + if value != value_next: + break + partnum += 1 + + return partnum + + + def _build_value_at_offset(self, value, offset, start, length): + """ + Return a couple. The first element is the memory Expression representing + the value at @offset, the second is its value. The value is truncated + at byte @start with @length + + @value: Expression to truncate + @offset: offset in bytes of the variable (integer) + @start: value's byte offset (integer) + @length: length in bytes (integer) + """ + + ptr = self.offset_to_ptr(self.base, offset) + size = length * 8 + if start == 0 and size == value.size: + result = value + else: + result = self.expr_simp(value[start * 8: start * 8 + size]) + + return ExprMem(ptr, size), result + + + def memory(self): + """ + Iterate on stored memory/values + + The goal here is to group entities. + + Consider the following state: + EAX + 0x10 = (0, EDX) + EAX + 0x11 = (1, EDX) + EAX + 0x12 = (2, EDX) + EAX + 0x13 = (3, EDX) + + The function should return: + @32[EAX + 0x10] = EDX + """ + + if not self._offset_to_expr: + raise StopIteration + known_offsets = self._offset_to_expr.keys() + known_offsets.sort() + index = 0 + # Test if the first element is the continuation of the last byte. If + # yes, merge and output it first. + min_int = 0 + max_int = (1 << self.base.size) - 1 + limit_index = len(known_offsets) + + first_element = None + # Special case where a variable spreads on max_int/min_int + if known_offsets[0] == min_int and known_offsets[-1] == max_int: + min_offset, max_offset = known_offsets[0], known_offsets[-1] + min_byte_index, min_value = self._offset_to_expr[min_offset] + max_byte_index, max_value = self._offset_to_expr[max_offset] + if min_value == max_value and max_byte_index + 1 == min_byte_index: + # Look for current variable start + partnum_before = self._get_variable_parts(len(known_offsets) - 1, known_offsets, False) + # Look for current variable end + partnum_after = self._get_variable_parts(0, known_offsets) + + partnum = partnum_before + partnum_after + offset = known_offsets[-partnum_before] + index_value, value = self._offset_to_expr[offset] + + mem, result = self._build_value_at_offset(value, offset, index_value, partnum) + first_element = mem, result + index = partnum_after + limit_index = len(known_offsets) - partnum_before + + # Special cases are done, walk and merge variables + while index < limit_index: + offset = known_offsets[index] + index_value, value = self._offset_to_expr[offset] + partnum = self._get_variable_parts(index, known_offsets) + mem, result = self._build_value_at_offset(value, offset, index_value, partnum) + yield mem, result + index += partnum + + if first_element is not None: + yield first_element + + def dump(self): + """Display MemArray content""" + for mem, value in self.memory(): + print "%s = %s" % (mem, value) + + +class MemSparse(object): + """Link a symbolic memory pointer to its MemArray. + + For each symbolic memory object, this object will extract the memory pointer + *ptr*. It then splits *ptr* into a symbolic and an integer part. For + example, the memory @[ESP+4] will give ESP+4 for *ptr*. *ptr* is then split + into its base ESP and its offset 4. Each symbolic base address uses a + different MemArray. + + Example: + - @32[EAX+EBX] + - @8[EAX+EBX+0x100] + Will be stored in the same MemArray with a EAX+EBX base + + """ + + def __init__(self, addrsize, expr_simp=expr_simp): + """ + @addrsize: size (in bits) of the addresses manipulated by the MemSparse + @expr_simp: an ExpressionSimplifier instance + """ + self.addrsize = addrsize + self.expr_simp = expr_simp + self.base_to_memarray = {} + + def __contains__(self, expr): + """ + Return True if the whole @expr is present + For partial check, use 'contains_partial' + """ + if not expr.is_mem(): + return False + ptr = expr.arg + base, offset = get_expr_base_offset(ptr) + memarray = self.base_to_memarray.get(base, None) + if memarray is None: + return False + for i in xrange(expr.size / 8): + if offset + i not in memarray: + return False + return True + + def contains_partial(self, expr): + """ + Return True if a part of @expr is present in memory + """ + if not expr.is_mem(): + return False + ptr = expr.arg + base, offset = get_expr_base_offset(ptr) + memarray = self.base_to_memarray.get(base, None) + if memarray is None: + return False + for i in xrange(expr.size / 8): + if offset + i in memarray: + return True + return False + + def clear(self): + """Reset the current object content""" + self.base_to_memarray.clear() + + def copy(self): + """Copy the current object instance""" + base_to_memarray = {} + for base, memarray in self.base_to_memarray.iteritems(): + base_to_memarray[base] = memarray.copy() + obj = MemSparse(self.addrsize, self.expr_simp) + obj.base_to_memarray = base_to_memarray + return obj + + def __delitem__(self, expr): + """ + Delete a value @expr *fully* present in memory + For partial delete, use delete_partial + """ + ptr = expr.arg + base, offset = get_expr_base_offset(ptr) + memarray = self.base_to_memarray.get(base, None) + if memarray is None: + raise KeyError + # Check if whole entity is in the MemArray before deleting it + for i in xrange(expr.size / 8): + if (offset + i) & memarray.mask not in memarray: + raise KeyError + for i in xrange(expr.size / 8): + del memarray[(offset + i) & memarray.mask] + + def delete_partial(self, expr): + """ + Delete @expr from memory. Skip parts of @expr which are not present in + memory. + """ + ptr = expr.arg + base, offset = get_expr_base_offset(ptr) + memarray = self.base_to_memarray.get(base, None) + if memarray is None: + raise KeyError + # Check if whole entity is in the MemArray before deleting it + for i in xrange(expr.size / 8): + real_offset = (offset + i) & memarray.mask + if real_offset in memarray: + del memarray[real_offset] + + def read(self, ptr, size): + """ + Return the value associated with the Expr at address @ptr + @ptr: Expr representing the memory address + @size: memory size (in bits), byte aligned + """ + assert size % 8 == 0 + base, offset = get_expr_base_offset(ptr) + memarray = self.base_to_memarray.get(base, None) + if memarray is not None: + mems = memarray.read(offset, size) + ret = ExprCompose(*mems) + else: + ret = ExprMem(ptr, size) + return ret + + def write(self, ptr, expr): + """ + Update the corresponding Expr @expr at address @ptr + @ptr: Expr representing the memory address + @expr: Expr instance + """ + assert ptr.size == self.addrsize + base, offset = get_expr_base_offset(ptr) + memarray = self.base_to_memarray.get(base, None) + if memarray is None: + memarray = MemArray(base, self.expr_simp) + self.base_to_memarray[base] = memarray + memarray.write(offset, expr) + + def iteritems(self): + """Iterate on stored memory variables and their values.""" + for _, memarray in sorted(self.base_to_memarray.iteritems()): + for mem, value in memarray.memory(): + yield mem, value + + def items(self): + """Return stored memory variables and their values.""" + return list(self.iteritems()) + + def dump(self): + """Display MemSparse content""" + for mem, value in self.iteritems(): + print "%s = %s" % (mem, value) + + def __repr__(self): + out = [] + for _, memarray in sorted(self.base_to_memarray.iteritems()): + out.append(repr(memarray)) + return '\n'.join(out) + + +class SymbolMngr(object): + """Symbolic store manager (IDs and MEMs)""" + + def __init__(self, init=None, addrsize=None, expr_simp=expr_simp): + assert addrsize is not None + if init is None: + init = {} + self.addrsize = addrsize + self.expr_simp = expr_simp + self.symbols_id = {} + self.symbols_mem = MemSparse(addrsize, expr_simp) + self.mask = (1 << addrsize) - 1 + for expr, value in init.iteritems(): + self.write(expr, value) + + def __contains__(self, expr): + if expr.is_id(): + return self.symbols_id.__contains__(expr) + if expr.is_mem(): + return self.symbols_mem.__contains__(expr) + return False + + def __getitem__(self, expr): + return self.read(expr) + + def __setitem__(self, expr, value): + self.write(expr, value) + + def __delitem__(self, expr): + if expr.is_id(): + del self.symbols_id[expr] + elif expr.is_mem(): + del self.symbols_mem[expr] + else: + raise TypeError("Bad source expr") + + def copy(self): + """Copy object instance""" + obj = SymbolMngr(self, addrsize=self.addrsize, expr_simp=self.expr_simp) + return obj + + def clear(self): + """Forget every variables values""" + self.symbols_id.clear() + self.symbols_mem.clear() + + def read(self, src): + """ + Return the value corresponding to Expr @src + @src: ExprId or ExprMem instance + """ + if src.is_id(): + return self.symbols_id.get(src, src) + elif src.is_mem(): + # Only byte aligned accesses are supported for now + assert src.size % 8 == 0 + return self.symbols_mem.read(src.arg, src.size) + else: + raise TypeError("Bad source expr") + + def write(self, dst, src): + """ + Update @dst with @src expression + @dst: ExprId or ExprMem instance + @src: Expression instance + """ + assert dst.size == src.size + if dst.is_id(): + if dst == src: + if dst in self.symbols_id: + del self.symbols_id[dst] + else: + self.symbols_id[dst] = src + elif dst.is_mem(): + # Only byte aligned accesses are supported for now + assert dst.size % 8 == 0 + self.symbols_mem.write(dst.arg, src) + else: + raise TypeError("Bad destination expr") + + def dump(self, ids=True, mems=True): + """Display memory content""" + if ids: + for variable, value in self.ids(): + print '%s = %s' % (variable, value) + if mems: + for mem, value in self.memory(): + print '%s = %s' % (mem, value) + + def __repr__(self): + out = [] + for variable, value in self.iteritems(): + out.append('%s = %s' % (variable, value)) + return "\n".join(out) + + def iteritems(self): + """ExprId/ExprMem iteritems of the current state""" + for variable, value in self.ids(): + yield variable, value + for variable, value in self.memory(): + yield variable, value + + def items(self): + """Return variables/values of the current state""" + return list(self.iteritems()) + + def __iter__(self): + for expr, _ in self.iteritems(): + yield expr + + def ids(self): + """Iterate on variables and their values.""" + for expr, value in self.symbols_id.iteritems(): + yield expr, value + + def memory(self): + """Iterate on memory variables and their values.""" + for mem, value in self.symbols_mem.iteritems(): + yield mem, value + + def keys(self): + """Variables of the current state""" + return list(self) + + def get(self, expr, default=None): + """Deprecated version of read""" + warnings.warn('DEPRECATION WARNING: use "read(self, expr)" instead of get') + ret = self.read(expr) + if default is not None and ret == expr: + return default + return ret + + +def merge_ptr_read(known, ptrs): + """ + Merge common memory parts in a multiple byte memory. + @ptrs: memory bytes list + @known: ptrs' associated boolean for present/unpresent memory part in the + store + """ + assert known + out = [] + known.append(None) + ptrs.append(None) + last, value, size = known[0], ptrs[0], 8 + for index, part in enumerate(known[1:], 1): + if part == last: + size += 8 + else: + out.append((last, value, size)) + last, value, size = part, ptrs[index], 8 + return out + + class SymbolicExecutionEngine(object): """ Symbolic execution engine Allow IR code emulation in symbolic domain + + + Examples: + from miasm2.ir.symbexec import SymbolicExecutionEngine + from miasm2.ir.ir import AssignBlock + + ir_arch = ir_x86_32() + + init_state = { + ir_arch.arch.regs.EAX: ir_arch.arch.regs.EBX, + ExprMem(id_x+ExprInt(0x10, 32), 32): id_a, + } + + sb_exec = SymbolicExecutionEngine(ir_arch, init_state) + + >>> sb_exec.dump() + EAX = a + @32[x + 0x10] = a + >>> sb_exec.dump(mems=False) + EAX = a + + >>> print sb_exec.eval_expr(ir_arch.arch.regs.EAX + ir_arch.arch.regs.ECX) + EBX + ECX + + Inspecting state: + - dump + - modified + State manipulation: + - '.state' (rw) + + Evaluation (read only): + - eval_expr + - eval_assignblk + Evaluation with state update: + - eval_updt_expr + - eval_updt_assignblk + - eval_updt_irblock + + Start a symbolic execution based on provisioned '.ir_arch' blocks: + - run_block_at + - run_at """ StateEngine = SymbolicState @@ -176,90 +806,31 @@ class SymbolicExecutionEngine(object): func_write=None, sb_expr_simp=expr_simp): - self.symbols = SymbolMngr() + self.expr_to_visitor = { + ExprInt: self.eval_exprint, + ExprId: self.eval_exprid, + ExprMem: self.eval_exprmem, + ExprSlice: self.eval_exprslice, + ExprCond: self.eval_exprcond, + ExprOp: self.eval_exprop, + ExprCompose: self.eval_exprcompose, + } + + self.symbols = SymbolMngr(addrsize=ir_arch.addrsize, expr_simp=expr_simp) + for dst, src in state.iteritems(): - self.symbols[dst] = src + self.symbols.write(dst, src) + + if func_read: + warnings.warn('DEPRECATION WARNING: override function "mem_read(self, expr)" instead of func_read') + if func_write: + warnings.warn('DEPRECATION WARNING: override function "mem_write(self, dsr, src)" instead of func_write') self.func_read = func_read self.func_write = func_write self.ir_arch = ir_arch self.expr_simp = sb_expr_simp - def find_mem_by_addr(self, expr): - """ - Return memory keys with pointer equal to @expr - @expr: address of the searched memory variable - """ - if expr in self.symbols.symbols_mem: - return self.symbols.symbols_mem[expr][0] - return None - - def get_mem_state(self, expr): - """ - Evaluate the @expr memory in the current state using @cache - @expr: the memory key - """ - ptr, size = expr.arg, expr.size - ret = self.find_mem_by_addr(ptr) - if not ret: - overlaps = self.get_mem_overlapping(expr) - if not overlaps: - if self.func_read and ptr.is_int(): - expr = self.func_read(expr) - return expr - - out = [] - off_base = 0 - for off, mem in overlaps: - if off >= 0: - new_size = min(size - off * 8, mem.size) - tmp = self.expr_simp(self.symbols[mem][0:new_size]) - out.append((tmp, off_base, off_base + new_size)) - off_base += new_size - else: - new_size = min(size - off * 8, mem.size) - tmp = self.expr_simp(self.symbols[mem][-off * 8:new_size]) - new_off_base = off_base + new_size + off * 8 - out.append((tmp, off_base, new_off_base)) - off_base = new_off_base - - missing_slice = self.rest_slice(out, 0, size) - for slice_start, slice_stop in missing_slice: - ptr = self.expr_simp(ptr + m2_expr.ExprInt(slice_start / 8, ptr.size)) - mem = m2_expr.ExprMem(ptr, slice_stop - slice_start) - if self.func_read and ptr.is_int(): - mem = self.func_read(mem) - out.append((mem, slice_start, slice_stop)) - out.sort(key=lambda x: x[1]) - args = [expr for (expr, _, _) in out] - ret = self.expr_simp(m2_expr.ExprCompose(*args)[:size]) - return ret - - # bigger lookup - if size > ret.size: - rest = size - out = [] - while rest: - mem = self.find_mem_by_addr(ptr) - if mem is None: - mem = m2_expr.ExprMem(ptr, 8) - if self.func_read and ptr.is_int(): - value = self.func_read(mem) - else: - value = mem - elif rest >= mem.size: - value = self.symbols[mem] - else: - value = self.symbols[mem][:rest] - out.append(value) - rest -= value.size - ptr = self.expr_simp(ptr + m2_expr.ExprInt(mem.size / 8, ptr.size)) - ret = self.expr_simp(m2_expr.ExprCompose(*out)) - return ret - # part lookup - ret = self.expr_simp(self.symbols[ret][:size]) - return ret - def get_state(self): """Return the current state of the SymbolicEngine""" state = self.StateEngine(dict(self.symbols)) @@ -269,214 +840,161 @@ class SymbolicExecutionEngine(object): """Restaure the @state of the engine @state: StateEngine instance """ - self.symbols = SymbolMngr() + self.symbols = SymbolMngr(addrsize=self.ir_arch.addrsize, expr_simp=self.expr_simp) for dst, src in dict(state).iteritems(): self.symbols[dst] = src - def apply_expr_on_state_visit_cache(self, expr, state, cache, level=0): + state = property(get_state, set_state) + + def eval_expr_visitor(self, expr, cache=None): """ - Deep First evaluate nodes: - 1. evaluate node's sons - 2. simplify + [DEV]: Override to change the behavior of an Expr evaluation. + This function recursively applies 'eval_expr*' to @expr. + This function uses @cache to speedup re-evaluation of expression. """ + if cache is None: + cache = {} - expr = self.expr_simp(expr) + ret = cache.get(expr, None) + if ret is not None: + return ret - #print '\t'*level, "Eval:", expr - if expr in cache: - ret = cache[expr] - #print "In cache!", ret - elif expr.is_int(): - return expr - elif expr.is_id(): - if isinstance(expr.name, asmblock.AsmLabel) and expr.name.offset is not None: - ret = m2_expr.ExprInt(expr.name.offset, expr.size) - else: - ret = state.get(expr, expr) - elif expr.is_mem(): - ptr = self.apply_expr_on_state_visit_cache(expr.arg, state, cache, level+1) - ret = m2_expr.ExprMem(ptr, expr.size) - ret = self.get_mem_state(ret) - assert expr.size == ret.size - elif expr.is_cond(): - cond = self.apply_expr_on_state_visit_cache(expr.cond, state, cache, level+1) - src1 = self.apply_expr_on_state_visit_cache(expr.src1, state, cache, level+1) - src2 = self.apply_expr_on_state_visit_cache(expr.src2, state, cache, level+1) - ret = m2_expr.ExprCond(cond, src1, src2) - elif expr.is_slice(): - arg = self.apply_expr_on_state_visit_cache(expr.arg, state, cache, level+1) - ret = m2_expr.ExprSlice(arg, expr.start, expr.stop) - elif expr.is_op(): - args = [] - for oarg in expr.args: - arg = self.apply_expr_on_state_visit_cache(oarg, state, cache, level+1) - assert oarg.size == arg.size - args.append(arg) - ret = m2_expr.ExprOp(expr.op, *args) - elif expr.is_compose(): - args = [] - for arg in expr.args: - args.append(self.apply_expr_on_state_visit_cache(arg, state, cache, level+1)) - ret = m2_expr.ExprCompose(*args) - else: + new_expr = self.expr_simp(expr) + ret = cache.get(new_expr, None) + if ret is not None: + return ret + + func = self.expr_to_visitor.get(new_expr.__class__, None) + if func is None: raise TypeError("Unknown expr type") - #print '\t'*level, "Result", ret + + ret = func(new_expr, cache=cache) ret = self.expr_simp(ret) - #print '\t'*level, "Result simpl", ret + assert ret is not None - assert expr.size == ret.size cache[expr] = ret + cache[new_expr] = ret return ret - def apply_expr_on_state(self, expr, cache): - if cache is None: - cache = {} - ret = self.apply_expr_on_state_visit_cache(expr, self.symbols, cache) + def eval_exprint(self, expr, **kwargs): + """[DEV]: Evaluate an ExprInt using the current state""" + return expr + + def eval_exprid(self, expr, **kwargs): + """[DEV]: Evaluate an ExprId using the current state""" + if isinstance(expr.name, asmblock.AsmLabel) and expr.name.offset is not None: + ret = ExprInt(expr.name.offset, expr.size) + else: + ret = self.symbols.read(expr) return ret - def eval_expr(self, expr, eval_cache=None): - return self.apply_expr_on_state(expr, eval_cache) + def eval_exprmem(self, expr, **kwargs): + """[DEV]: Evaluate an ExprMem using the current state + This function first evaluate the memory pointer value. + Override 'mem_read' to modify the effective memory accesses + """ + ptr = self.eval_expr_visitor(expr.arg, **kwargs) + mem = ExprMem(ptr, expr.size) + ret = self.mem_read(mem) + return ret - def modified_regs(self, init_state=None): - if init_state is None: - init_state = self.ir_arch.arch.regs.regs_init - ids = self.symbols.symbols_id.keys() - ids.sort() - for i in ids: - if i in init_state and \ - i in self.symbols.symbols_id and \ - self.symbols.symbols_id[i] == init_state[i]: - continue - yield i + def eval_exprcond(self, expr, **kwargs): + """[DEV]: Evaluate an ExprCond using the current state""" + cond = self.eval_expr_visitor(expr.cond, **kwargs) + src1 = self.eval_expr_visitor(expr.src1, **kwargs) + src2 = self.eval_expr_visitor(expr.src2, **kwargs) + ret = ExprCond(cond, src1, src2) + return ret - def modified_mems(self, init_state=None): - if init_state is None: - init_state = self.ir_arch.arch.regs.regs_init - mems = self.symbols.symbols_mem.values() - mems.sort() - for mem, _ in mems: - if mem in init_state and \ - mem in self.symbols.symbols_mem and \ - self.symbols.symbols_mem[mem] == init_state[mem]: - continue - yield mem + def eval_exprslice(self, expr, **kwargs): + """[DEV]: Evaluate an ExprSlice using the current state""" + arg = self.eval_expr_visitor(expr.arg, **kwargs) + ret = ExprSlice(arg, expr.start, expr.stop) + return ret - def modified(self, init_state=None): - for reg in self.modified_regs(init_state): - yield reg - for mem in self.modified_mems(init_state): - yield mem + def eval_exprop(self, expr, **kwargs): + """[DEV]: Evaluate an ExprOp using the current state""" + args = [] + for oarg in expr.args: + arg = self.eval_expr_visitor(oarg, **kwargs) + args.append(arg) + ret = ExprOp(expr.op, *args) + return ret - def dump_id(self): - """ - Dump modififed registers symbols only - """ - ids = self.symbols.symbols_id.keys() - ids.sort() - for expr in ids: - if (expr in self.ir_arch.arch.regs.regs_init and - expr in self.symbols.symbols_id and - self.symbols.symbols_id[expr] == self.ir_arch.arch.regs.regs_init[expr]): - continue - print expr, "=", self.symbols.symbols_id[expr] + def eval_exprcompose(self, expr, **kwargs): + """[DEV]: Evaluate an ExprCompose using the current state""" + args = [] + for arg in expr.args: + args.append(self.eval_expr_visitor(arg, **kwargs)) + ret = ExprCompose(*args) + return ret - def dump_mem(self): + def eval_expr(self, expr, eval_cache=None): """ - Dump modififed memory symbols + Evaluate @expr + @expr: Expresion instance to evaluate + @cache: None or dictionary linking variables to their values """ - mems = self.symbols.symbols_mem.values() - mems.sort() - for mem, value in mems: - print mem, value + if eval_cache is None: + eval_cache = {} + ret = self.eval_expr_visitor(expr, cache=eval_cache) + assert ret is not None + return ret - def rest_slice(self, slices, start, stop): + def modified(self, init_state=None, ids=True, mems=True): """ - Return the complementary slices of @slices in the range @start, @stop - @slices: base slices - @start, @stop: interval range + Return the modified variables. + @init_state: a base dictionary linking variables to their initial values + to diff. Can be None. + @ids: track ids only + @mems: track mems only """ - out = [] - last = start - for _, slice_start, slice_stop in slices: - if slice_start == last: - last = slice_stop - continue - out.append((last, slice_start)) - last = slice_stop - if last != stop: - out.append((slice_stop, stop)) - return out - - def substract_mems(self, arg1, arg2): + if init_state is None: + init_state = self.ir_arch.arch.regs.regs_init + if ids: + for variable, value in self.symbols.symbols_id.iteritems(): + if variable in init_state and init_state[variable] == value: + continue + yield variable, value + if mems: + for mem, value in self.symbols.memory(): + if mem in init_state and init_state[mem] == value: + continue + yield mem, value + + def dump(self, ids=True, mems=True): """ - Return the remaining memory areas of @arg1 - @arg2 - @arg1, @arg2: ExprMem + Display modififed variables + @ids: display modified ids + @mems: display modified memory """ - ptr_diff = self.expr_simp(arg2.arg - arg1.arg) - ptr_diff = int(int32(ptr_diff.arg)) - - zone1 = interval([(0, arg1.size/8-1)]) - zone2 = interval([(ptr_diff, ptr_diff + arg2.size/8-1)]) - zones = zone1 - zone2 - - out = [] - for start, stop in zones: - ptr = arg1.arg + m2_expr.ExprInt(start, arg1.arg.size) - ptr = self.expr_simp(ptr) - value = self.expr_simp(self.symbols[arg1][start*8:(stop+1)*8]) - mem = m2_expr.ExprMem(ptr, (stop - start + 1)*8) - assert mem.size == value.size - out.append((mem, value)) - - return out + for variable, value in self.modified(None, ids, mems): + print "%-18s" % variable, "=", "%s" % value - def get_mem_overlapping(self, expr): - """ - Gives mem stored overlapping memory in @expr - Hypothesis: Max mem size is 64 bytes, compute all reachable addresses - @expr: target memory + def eval_assignblk(self, assignblk): """ + Evaluate AssignBlock using the current state - overlaps = [] - base_ptr = self.expr_simp(expr.arg) - for i in xrange(-7, expr.size / 8): - new_ptr = base_ptr + m2_expr.ExprInt(i, expr.arg.size) - new_ptr = self.expr_simp(new_ptr) - - mem, origin = self.symbols.symbols_mem.get(new_ptr, (None, None)) - if mem is None: - continue - - ptr_diff = -i - if ptr_diff >= origin.size / 8: - # access is too small to overlap the memory target - continue - overlaps.append((i, mem)) + Returns a dictionary containing modified keys associated to their values - return overlaps - - def eval_ir_expr(self, assignblk): - """ - Evaluate AssignBlock on the current state @assignblk: AssignBlock instance """ pool_out = {} eval_cache = {} for dst, src in assignblk.iteritems(): src = self.eval_expr(src, eval_cache) - if isinstance(dst, m2_expr.ExprMem): + if dst.is_mem(): ptr = self.eval_expr(dst.arg, eval_cache) - # test if mem lookup is known - tmp = m2_expr.ExprMem(ptr, dst.size) + # Test if mem lookup is known + tmp = ExprMem(ptr, dst.size) pool_out[tmp] = src - - elif isinstance(dst, m2_expr.ExprId): + elif dst.is_id(): pool_out[dst] = src else: raise ValueError("Unknown destination type", str(dst)) - return pool_out.iteritems() + return pool_out def apply_change(self, dst, src): """ @@ -484,41 +1002,25 @@ class SymbolicExecutionEngine(object): @dst: Expr, destination @src: Expr, source """ - if isinstance(dst, m2_expr.ExprMem): - mem_overlap = self.get_mem_overlapping(dst) - for _, base in mem_overlap: - diff_mem = self.substract_mems(base, dst) - del self.symbols[base] - for new_mem, new_val in diff_mem: - self.symbols[new_mem] = new_val - src_o = self.expr_simp(src) - - # Force update. Ex: - # EBX += 1 (state: EBX = EBX+1) - # EBX -= 1 (state: EBX = EBX, must be updated) - self.symbols[dst] = src_o - if dst == src_o: - # Avoid useless X = X information - del self.symbols[dst] - if isinstance(dst, m2_expr.ExprMem): - if self.func_write and isinstance(dst.arg, m2_expr.ExprInt): - self.func_write(self, dst, src_o) - del self.symbols[dst] + if dst.is_mem(): + self.mem_write(dst, src) + else: + self.symbols.write(dst, src) - def eval_ir(self, assignblk): + def eval_updt_assignblk(self, assignblk): """ Apply an AssignBlock on the current state @assignblk: AssignBlock instance """ mem_dst = [] - src_dst = self.eval_ir_expr(assignblk) - for dst, src in src_dst: + dst_src = self.eval_assignblk(assignblk) + for dst, src in dst_src.iteritems(): self.apply_change(dst, src) - if isinstance(dst, m2_expr.ExprMem): + if dst.is_mem(): mem_dst.append(dst) return mem_dst - def emulbloc(self, irb, step=False): + def eval_updt_irblock(self, irb, step=False): """ Symbolic execution of the @irb on the current state @irb: irbloc instance @@ -530,35 +1032,38 @@ class SymbolicExecutionEngine(object): print 'Assignblk:' print assignblk print '_' * 80 - self.eval_ir(assignblk) + self.eval_updt_assignblk(assignblk) if step: - self.dump_id() - self.dump_mem() + self.dump(mems=False) + self.dump(ids=False) print '_' * 80 return self.eval_expr(self.ir_arch.IRDst) - def emul_ir_bloc(self, _, addr, step=False): - warnings.warn('DEPRECATION WARNING: use "emul_ir_block(self, addr, step=False)" instead of emul_ir_bloc') - return self.emul_ir_block(addr, step) - - def emul_ir_block(self, addr, step=False): + def run_block_at(self, addr, step=False): + """ + Symbolic execution of the block at @addr + @addr: address to execute (int or ExprInt or label) + @step: display intermediate steps + """ irblock = self.ir_arch.get_block(addr) if irblock is not None: - addr = self.emulbloc(irblock, step=step) + addr = self.eval_updt_irblock(irblock, step=step) return addr - def emul_ir_blocs(self, _, addr, lbl_stop=None, step=False): - warnings.warn('DEPRECATION WARNING: use "emul_ir_blocks(self, addr, lbl_stop=None, step=False):" instead of emul_ir_blocs') - return self.emul_ir_blocks(addr, lbl_stop, step) - - def emul_ir_blocks(self, addr, lbl_stop=None, step=False): + def run_at(self, addr, lbl_stop=None, step=False): + """ + Symbolic execution starting at @addr + @addr: address to execute (int or ExprInt or label) + @lbl_stop: AsmLabel to stop execution on + @step: display intermediate steps + """ while True: irblock = self.ir_arch.get_block(addr) if irblock is None: break if irblock.label == lbl_stop: break - addr = self.emulbloc(irblock, step=step) + addr = self.eval_updt_irblock(irblock, step=step) return addr def del_mem_above_stack(self, stack_ptr): @@ -568,32 +1073,177 @@ class SymbolicExecutionEngine(object): * pointer below current stack pointer """ stack_ptr = self.eval_expr(stack_ptr) - for mem_addr, (mem, _) in self.symbols.symbols_mem.items(): - diff = self.expr_simp(mem_addr - stack_ptr) - if not isinstance(diff, m2_expr.ExprInt): - continue - sign_bit = self.expr_simp(diff.msb()) - if sign_bit.arg == 1: - del self.symbols[mem] - - def apply_expr(self, expr): - """Evaluate @expr and apply side effect if needed (ie. if expr is an - assignment). Return the evaluated value""" + base, stk_offset = get_expr_base_offset(stack_ptr) + memarray = self.symbols.symbols_mem.base_to_memarray.get(base, None) + if memarray: + to_del = set() + for offset in memarray: + if ((offset - stk_offset) & int(stack_ptr.mask)) >> (stack_ptr.size - 1) != 0: + to_del.add(offset) + + for offset in to_del: + del memarray[offset] + + def eval_updt_expr(self, expr): + """ + Evaluate @expr and apply side effect if needed (ie. if expr is an + assignment). Return the evaluated value + """ # Update value if needed - if isinstance(expr, m2_expr.ExprAff): + if expr.is_aff(): ret = self.eval_expr(expr.src) - self.eval_ir(AssignBlock([expr])) + self.eval_updt_assignblk(AssignBlock([expr])) else: ret = self.eval_expr(expr) return ret + def _resolve_mem_parts(self, expr): + """For a given ExprMem @expr, get known/unknown parts from the store. + @expr: ExprMem instance + + Return a list of (known, value) where known is a bool representing if + the value has been resolved from the store or not. + """ + + # Extract known parts in symbols + assert expr.size % 8 == 0 + ptr = expr.arg + known = [] + ptrs = [] + for index in xrange(expr.size / 8): + offset = self.expr_simp(ptr + ExprInt(index, ptr.size)) + ptrs.append(offset) + mem = ExprMem(offset, 8) + known.append(mem in self.symbols) + + reads = merge_ptr_read(known, ptrs) + out = [] + for is_known, ptr_value, size in reads: + mem = ExprMem(ptr_value, size) + if is_known: + mem = self.symbols.read(mem) + out.append((is_known, mem)) + return out + + def mem_read(self, expr): + """ + [DEV]: Override to modify the effective memory reads + + Read symbolic value at ExprMem @expr + @expr: ExprMem + """ + + parts = self._resolve_mem_parts(expr) + + out = [] + for known, part in parts: + if not known and part.is_mem() and self.func_read is not None: + ret = self.func_read(part) + else: + ret = part + + out.append(ret) + ret = self.expr_simp(ExprCompose(*out)) + + assert ret.size == expr.size + return ret + + def mem_write(self, dst, src): + """ + [DEV]: Override to modify the effective memory writes + + Write symbolic value @src at ExprMem @dst + @dst: destination ExprMem + @src: source Expression + """ + if self.func_write is not None: + self.func_write(self, dst, src) + else: + self.symbols.write(dst, src) + + + # Deprecated methods + + def apply_expr_on_state(self, expr, cache): + """Deprecated version of eval_expr""" + warnings.warn('DEPRECATION WARNING: use "eval_expr" instead of apply_expr_on_state') + + if cache is None: + cache = {} + ret = self.eval_expr(expr, eval_cache=cache) + return ret + + def modified_mems(self, init_state=None): + """Deprecated version of modified(ids=False)""" + warnings.warn('DEPRECATION WARNING: use "modified(self, ids=False)" instead of modified_mems') + for mem in self.modified(init_state=init_state, ids=False): + yield mem + + def modified_regs(self, init_state=None): + """Deprecated version of modified(mems=False)""" + warnings.warn('DEPRECATION WARNING: use "modified(self, mems=False)" instead of modified_regs') + for reg in self.modified(init_state=init_state, mems=False): + yield reg + + def dump_id(self): + """Deprecated version of dump(mems=False)""" + warnings.warn('DEPRECATION WARNING: use "dump(self, mems=False)" instead of dump_id') + self.dump(mems=False) + + def dump_mem(self): + """Deprecated version of dump(ids=False)""" + warnings.warn('DEPRECATION WARNING: use "dump(self, ids=False)" instead of dump_mem') + self.dump(ids=False) + + def eval_ir_expr(self, assignblk): + """Deprecated version of eval_ir_expr(self, assignblk)""" + warnings.warn('DEPRECATION WARNING: use "eval_assignblk(self, assignblk)" instead of eval_ir_expr') + return self.eval_assignblk(assignblk).iteritems() + + def eval_ir(self, assignblk): + """Deprecated version of eval_updt_assignblk(self, assignblk)""" + warnings.warn('DEPRECATION WARNING: use "eval_assignblk(self, assignblk)" instead of eval_ir') + return self.eval_updt_assignblk(assignblk) + + def emulbloc(self, irb, step=False): + """Deprecated version of eval_updt_irblock(self, irb, step=False)""" + warnings.warn('DEPRECATION WARNING: use "eval_updt_irblock(self, irb, step=False)" instead of emulbloc') + return self.eval_updt_irblock(irb, step) + + def emul_ir_bloc(self, _, addr, step=False): + """Deprecated version of run_block_at""" + warnings.warn('DEPRECATION WARNING: use "run_block_at(self, addr, step=False)" instead of emul_ir_bloc') + return self.run_block_at(addr, step) + + def emul_ir_block(self, addr, step=False): + """Deprecated version of run_block_at""" + warnings.warn('DEPRECATION WARNING: use "run_block_at(self, addr, step=False)" instead of emul_ir_block') + return self.run_block_at(addr, step) + + def emul_ir_blocks(self, addr, lbl_stop=None, step=False): + """Deprecated version of run_at""" + warnings.warn('DEPRECATION WARNING: use "run_at(self, addr, lbl_stop=None, step=False):" instead of emul_ir_blocks') + return self.run_at(addr, lbl_stop, step) + + def emul_ir_blocs(self, _, addr, lbl_stop=None, step=False): + """Deprecated version of run_at""" + warnings.warn('DEPRECATION WARNING: use "run_at(self, addr, lbl_stop=None, step=False):" instead of emul_ir_blocs') + return self.run_at(addr, lbl_stop, step) + + def apply_expr(self, expr): + """Deprecated version of eval_updt_expr""" + warnings.warn('DEPRECATION WARNING: use "eval_updt_expr" instead of apply_expr') + return self.eval_updt_expr(expr) + def as_assignblock(self): """Return the current state as an AssignBlock""" - return AssignBlock({ - dst: self.symbols[dst] for dst in self.modified() - }) + warnings.warn('DEPRECATION WARNING: use "modified(ids=True, mems=True)" instead of as_assignblock') + out = [] + for dst, src in self.modified(ids=True, mems=True): + out.append((dst, src)) + return AssignBlock(dict(out)) class symbexec(SymbolicExecutionEngine): |