Add EXPLICIT fork policy (#2514)

Add a new fork policy called "EXPLICIT" that allows to specify arbitrary concrete values for the expression on whom we are forking.

* Add EXPLICIT fork policy allowing to provide our own concrete values

* Add feasibility checks for the EXPLICIT concretisation policy

* Add few type hints

* Don't pass kwargs to BaseException

* Only use lists in StateBase.concretize()

manticore/core/manticore.py

@@ -6,7 +6,7 @@
 import typing
 import random
 import weakref
-from typing import Callable
+from typing import Callable, List, Any, Optional
 
 from contextlib import contextmanager
 
@@ -434,7 +434,7 @@ def goto_snapshot(self):
     @sync
     @only_from_main_script
     def clear_snapshot(self):
-        """ Remove any saved states """
+        """Remove any saved states"""
         if self._snapshot:
             for state_id in self._snapshot:
                 self._remove(state_id)
@@ -443,7 +443,7 @@ def clear_snapshot(self):
     @sync
     @at_not_running
     def clear_terminated_states(self):
-        """ Remove all states from the terminated list """
+        """Remove all states from the terminated list"""
         terminated_states_ids = tuple(self._terminated_states)
         for state_id in terminated_states_ids:
             self._terminated_states.remove(state_id)
@@ -453,7 +453,7 @@ def clear_terminated_states(self):
     @sync
     @at_not_running
     def clear_ready_states(self):
-        """ Remove all states from the ready list """
+        """Remove all states from the ready list"""
         ready_states_ids = tuple(self._ready_states)
         for state_id in ready_states_ids:
             self._ready_states.remove(state_id)
@@ -480,7 +480,9 @@ def from_saved_state(cls, filename: str, *args, **kwargs):
 
         return cls(deserialized, *args, **kwargs)
 
-    def _fork(self, state, expression, policy="ALL", setstate=None):
+    def _fork(
+        self, state, expression, policy="ALL", setstate=None, values: Optional[List[Any]] = None
+    ):
         """
         Fork state on expression concretizations.
         Using policy build a list of solutions for expression.
@@ -510,7 +512,7 @@ def setstate(x, y):
                 pass
 
         # Find a set of solutions for expression
-        solutions = state.concretize(expression, policy)
+        solutions = state.concretize(expression, policy, explicit_values=values)
 
         if not solutions:
             raise ManticoreError("Forking on unfeasible constraint set")
@@ -626,7 +628,7 @@ def _put_state(self, state) -> int:
         return state_id
 
     def _get_state(self, wait=False) -> typing.Optional[StateBase]:
-        """ Dequeue a state form the READY list and add it to the BUSY list """
+        """Dequeue a state form the READY list and add it to the BUSY list"""
         with self._lock:
             # If wait is true do the conditional wait for states
             if wait:
@@ -860,32 +862,32 @@ def all_states(self):
 
     @sync
     def count_states(self):
-        """ Total states count """
+        """Total states count"""
         return len(self._all_states)
 
     @sync
     def count_all_states(self):
-        """ Total states count """
+        """Total states count"""
         return self.count_states()
 
     @sync
     def count_ready_states(self):
-        """ Ready states count """
+        """Ready states count"""
         return len(self._ready_states)
 
     @sync
     def count_busy_states(self):
-        """ Busy states count """
+        """Busy states count"""
         return len(self._busy_states)
 
     @sync
     def count_killed_states(self):
-        """ Cancelled states count """
+        """Cancelled states count"""
         return len(self._killed_states)
 
     @sync
     def count_terminated_states(self):
-        """ Terminated states count """
+        """Terminated states count"""
         return len(self._terminated_states)
 
     def generate_testcase(self, state, message: str = "test", name: str = "test") -> Testcase:
@@ -978,7 +980,7 @@ def unregister_plugin(self, plugin: typing.Union[str, Plugin]):
         plugin_inst.manticore = None
 
     def subscribe(self, name, callback):
-        """ Register a callback to an event"""
+        """Register a callback to an event"""
         from types import MethodType
 
         if not isinstance(callback, MethodType):
@@ -1045,7 +1047,7 @@ def locked_context(self, key=None, value_type=list):
 
     @sync
     def wait(self, condition):
-        """ Waits for the condition callable to return True """
+        """Waits for the condition callable to return True"""
         self._lock.wait_for(condition)
 
     @sync
@@ -1065,15 +1067,15 @@ def terminate(self):
 
     @sync
     def is_running(self):
-        """ True if workers are exploring BUSY states or waiting for READY states """
+        """True if workers are exploring BUSY states or waiting for READY states"""
         # If there are still states in the BUSY list then the STOP/KILL event
         # was not yet answered
         # We know that BUSY states can only decrease after a stop is requested
         return self._running.value
 
     @sync
     def is_killed(self):
-        """ True if workers are killed. It is safe to join them """
+        """True if workers are killed. It is safe to join them"""
         # If there are still states in the BUSY list then the STOP/KILL event
         # was not yet answered
         # We know that BUSY states can only decrease after a kill is requested
@@ -1277,5 +1279,5 @@ def register_daemon(self, callback: typing.Callable):
         self._daemon_callbacks.append(callback)
 
     def pretty_print_states(self, *_args):
-        """ Calls pretty_print_state_descriptors on the current set of state descriptors """
+        """Calls pretty_print_state_descriptors on the current set of state descriptors"""
         pretty_print_state_descriptors(self.introspect())

manticore/core/state.py

@@ -1,7 +1,7 @@
 import copy
 import logging
 
-from typing import List, Tuple, Sequence
+from typing import List, Tuple, Sequence, Optional, Any
 
 from .smtlib import solver, Bool, issymbolic, BitVecConstant
 from .smtlib.expression import Expression
@@ -21,11 +21,11 @@
 
 
 class StateException(Exception):
-    """ All state related exceptions """
+    """All state related exceptions"""
 
 
 class TerminateState(StateException):
-    """ Terminates current state exploration """
+    """Terminates current state exploration"""
 
     def __init__(self, message, testcase=False):
         super().__init__(message)
@@ -51,9 +51,27 @@ class Concretize(StateException):
 
     """
 
-    _ValidPolicies = ["MIN", "MAX", "MINMAX", "ALL", "SAMPLED", "ONE", "PESSIMISTIC", "OPTIMISTIC"]
-
-    def __init__(self, message, expression, setstate=None, policy=None, **kwargs):
+    _ValidPolicies = [
+        "MIN",
+        "MAX",
+        "MINMAX",
+        "ALL",
+        "SAMPLED",
+        "ONE",
+        "PESSIMISTIC",
+        "OPTIMISTIC",
+        "EXPLICIT",
+    ]
+
+    def __init__(
+        self,
+        message,
+        expression,
+        setstate=None,
+        policy=None,
+        values: Optional[List[Any]] = None,
+        **kwargs,
+    ):
         if policy is None:
             policy = "ALL"
         if policy not in self._ValidPolicies:
@@ -63,8 +81,9 @@ def __init__(self, message, expression, setstate=None, policy=None, **kwargs):
         self.expression = expression
         self.setstate = setstate
         self.policy = policy
+        self.values = values
         self.message = f"Concretize: {message} (Policy: {policy})"
-        super().__init__(**kwargs)
+        super().__init__()
 
 
 class SerializeState(Concretize):
@@ -365,7 +384,7 @@ def new_symbolic_value(self, nbits, label=None, taint=frozenset()):
         self._input_symbols.append(expr)
         return expr
 
-    def concretize(self, symbolic, policy, maxcount=7):
+    def concretize(self, symbolic, policy, maxcount=7, explicit_values: Optional[List[Any]] = None):
         """This finds a set of solutions for symbolic using policy.
 
         This limits the number of solutions returned to `maxcount` to avoid
@@ -376,13 +395,13 @@ def concretize(self, symbolic, policy, maxcount=7):
         assert self.constraints == self.platform.constraints
         symbolic = self.migrate_expression(symbolic)
 
-        vals = []
+        vals: List[Any] = []
         if policy == "MINMAX":
-            vals = self._solver.minmax(self._constraints, symbolic)
+            vals = list(self._solver.minmax(self._constraints, symbolic))
         elif policy == "MAX":
-            vals = (self._solver.max(self._constraints, symbolic),)
+            vals = [self._solver.max(self._constraints, symbolic)]
         elif policy == "MIN":
-            vals = (self._solver.min(self._constraints, symbolic),)
+            vals = [self._solver.min(self._constraints, symbolic)]
         elif policy == "SAMPLED":
             m, M = self._solver.minmax(self._constraints, symbolic)
             vals += [m, M]
@@ -404,17 +423,24 @@ def concretize(self, symbolic, policy, maxcount=7):
         elif policy == "OPTIMISTIC":
             logger.info("Optimistic case when forking")
             if self._solver.can_be_true(self._constraints, symbolic):
-                vals = (True,)
+                vals = [True]
             else:
                 # We assume the path constraint was feasible to begin with
-                vals = (False,)
+                vals = [False]
         elif policy == "PESSIMISTIC":
             logger.info("Pessimistic case when forking")
             if self._solver.can_be_true(self._constraints, symbolic == False):
-                vals = (False,)
+                vals = [False]
             else:
                 # We assume the path constraint was feasible to begin with
-                vals = (True,)
+                vals = [True]
+        elif policy == "EXPLICIT":
+            if explicit_values:
+                for val in explicit_values:
+                    if self._solver.can_be_true(self._constraints, val == symbolic):
+                        vals.append(val)
+                    if len(vals) >= maxcount:
+                        break
         else:
             assert policy == "ALL"
             vals = self._solver.get_all_values(

manticore/core/worker.py

@@ -155,7 +155,7 @@ def run(self, *args):
                         # Though, normally fork() saves the spawned childs,
                         # returns a None and let _get_state choose what to explore
                         # next
-                        m._fork(current_state, exc.expression, exc.policy, exc.setstate)
+                        m._fork(current_state, exc.expression, exc.policy, exc.setstate, exc.values)
                         current_state = None
 
                     except TerminateState as exc: