concretor_angr.py

from angr import sim_options as options
from angr.errors import SimUnsatError, SimMemoryError, SimMemoryLimitError, SimMemoryAddressError, SimMergeError
from angr.storage.memory import SimMemory, DUMMY_SYMBOLIC_READ_VALUE

import angr
import logging

l = logging.getLogger("concretor_angr")


def patched_load(self, dst, size, condition=None, fallback=None,
        inspect=True, events=True, ret_on_segv=False):
    if self.state.solver.symbolic(size):
        l.warning("Concretizing symbolic length. Much sad; think about implementing.")

    # for now, we always load the maximum size
    _,max_size = self._resolve_size_range(size)
    if options.ABSTRACT_MEMORY not in self.state.options and self.state.solver.symbolic(size):
        self.state.add_constraints(size == max_size, action=True)

    if max_size == 0:
        self.state.history.add_event('memory_limit', message="0-length read")

    size = max_size
    if self.state.solver.symbolic(dst) and options.AVOID_MULTIVALUED_READS in self.state.options:
        return [ ], self.get_unconstrained_bytes("symbolic_read_unconstrained", size*self.state.arch.byte_width), [ ]

    # get a concrete set of read addresses
    try:
        addrs = self.concretize_read_addr(dst)
    except SimMemoryError:
        if options.CONSERVATIVE_READ_STRATEGY in self.state.options:
            return [ ], self.get_unconstrained_bytes(
                "symbolic_read_unconstrained", size*self.state.arch.byte_width
            ), [ ]
        else:
            raise
    
    # concretize on read
    if "concretor" in self.state.plugins and self.state.concretor.active:
        read_value = self._read_from(addrs[0], size, inspect=inspect, events=events)
        is_sym = read_value.symbolic
        read_value = self.state.solver.BVV(self.state.solver.eval(read_value), size*8)
        if is_sym:
            l.warning("%s forcing concretization of 0x%x:%d to '%s'" % (repr(self.state.regs.pc), addrs[0], size, repr(read_value)))
        return addrs[:1], read_value, []
    
    constraint_options = [ ]

    if len(addrs) == 1:
        # It's not an conditional reaed
        constraint_options.append(dst == addrs[0])
        read_value = self._read_from(addrs[0], size, inspect=inspect, events=events)
    else:
        read_value = DUMMY_SYMBOLIC_READ_VALUE  # it's a sentinel value and should never be touched

        for a in addrs:
            read_value = self.state.solver.If(dst == a, self._read_from(a, size, inspect=inspect, events=events),
                                          read_value)
            constraint_options.append(dst == a)

    if len(constraint_options) > 1:
        load_constraint = [ self.state.solver.Or(*constraint_options) ]
    elif not self.state.solver.symbolic(constraint_options[0]):
        load_constraint = [ ]
    else:
        load_constraint = [ constraint_options[0] ]

    if condition is not None and fallback is not None:
        read_value = self.state.solver.If(condition, read_value, fallback)
        load_constraint = [ self.state.solver.Or(self.state.solver.And(condition, *load_constraint), self.state.solver.Not(condition)) ]

    return addrs, read_value, load_constraint



class ConcretorPlugin(angr.SimStatePlugin):
    
    def __init__(self, active=False, store_concretized=False):
        super(ConcretorPlugin, self).__init__()
        self.active = active
        self.store_concretized = store_concretized # TODO

    @angr.SimStatePlugin.memo
    def copy(self, memo):
        return ConcretorPlugin(self.active, self.store_concretized)


class ConcretorHook(angr.SimProcedure):
    IS_FUNCTION = True
    ARGS_MISMATCH = True
    local_vars = ('active_prev',)
    
    def run(self, addr=None, store_concretized=False):
        if "concretor" not in self.state.plugins:
            self.state.register_plugin("concretor", ConcretorPlugin(store_concretized=store_concretized))
        
        self.active_prev = self.state.concretor.active
        self.state.concretor.active = True
        self.call(addr, [], "retfn")
    
    def retfn(self, addr=None):
        self.state.concretor.active = False


def set_concrete_import(project, sym):
    addr = project.loader.main_object.plt[sym]
    project.hook_symbol(addr, ConcretorHook(addr=project.loader.find_symbol(sym).rebased_addr))

def set_concrete_missing_simprocs(project):
    simprocs_names = map(lambda x: x.display_name, project._sim_procedures.values())
    for sym in filter(lambda x: x not in simprocs_names, project.loader.main_object.plt):
        l.info("setting concrete import %s" % sym)
        set_concrete_import(project, sym)


angr.state_plugins.symbolic_memory.SimSymbolicMemory._load = patched_load