Merge pull request #897 from ConsenSys/enhance/604

Reachability Check

mythril/analysis/solver.py

@@ -13,16 +13,19 @@ import logging
 log = logging.getLogger(__name__)
 
 
-def get_model(constraints, minimize=(), maximize=()):
+def get_model(constraints, minimize=(), maximize=(), enforce_execution_time=True):
     """
 
     :param constraints:
     :param minimize:
     :param maximize:
+    :param enforce_execution_time: Bool variable which enforces --execution-timeout's time
     :return:
     """
     s = Optimize()
-    timeout = min(100000, time_handler.time_remaining() - 500)
+    timeout = 100000
+    if enforce_execution_time:
+        timeout = min(timeout, time_handler.time_remaining() - 500)
         if timeout <= 0:
             raise UnsatError
     s.set_timeout(timeout)

mythril/laser/ethereum/instructions.py

@@ -1527,9 +1527,18 @@ class Instruction:
             simplify(Not(condition)) if isinstance(condition, Bool) else condition == 0
         )
         negated.simplify()
-        if (type(negated) == bool and negated) or (
+        # True case
+        condi = simplify(condition) if isinstance(condition, Bool) else condition != 0
+        condi.simplify()
+
+        negated_cond = (type(negated) == bool and negated) or (
             isinstance(negated, Bool) and not is_false(negated)
-        ):
+        )
+        positive_cond = (type(condi) == bool and condi) or (
+            isinstance(condi, Bool) and not is_false(condi)
+        )
+
+        if negated_cond:
             new_state = copy(global_state)
             # add JUMPI gas cost
             new_state.mstate.min_gas_used += min_gas
@@ -1547,18 +1556,15 @@ class Instruction:
 
         # Get jump destination
         index = util.get_instruction_index(disassembly.instruction_list, jump_addr)
+
         if not index:
             log.debug("Invalid jump destination: " + str(jump_addr))
             return states
 
         instr = disassembly.instruction_list[index]
 
-        condi = simplify(condition) if isinstance(condition, Bool) else condition != 0
-        condi.simplify()
         if instr["opcode"] == "JUMPDEST":
-            if (type(condi) == bool and condi) or (
-                isinstance(condi, Bool) and not is_false(condi)
-            ):
+            if positive_cond:
                 new_state = copy(global_state)
                 # add JUMPI gas cost
                 new_state.mstate.min_gas_used += min_gas
@@ -1571,7 +1577,6 @@ class Instruction:
                 states.append(new_state)
             else:
                 log.debug("Pruned unreachable states.")
-        del global_state
         return states
 
     @StateTransition()

mythril/laser/ethereum/state/constraints.py

@@ -1,77 +1,100 @@
 """This module contains the class used to represent state-change constraints in
 the call graph."""
 
+from mythril.laser.smt import Solver, Bool
+
+from typing import Iterable, List, Optional
+from z3 import unsat
+
 
 class Constraints(list):
     """This class should maintain a solver and it's constraints, This class
     tries to make the Constraints() object as a simple list of constraints with
     some background processing.
 
-    TODO: add the solver to this class after callback refactor
     """
 
-    def __init__(self, constraint_list=None, solver=None, possibility=None):
+    def __init__(
+        self,
+        constraint_list: Optional[List[Bool]] = None,
+        is_possible: Optional[bool] = None,
+    ) -> None:
         """
 
-        :param constraint_list:
-        :param solver:
-        :param possibility:
+        :param constraint_list: List of constraints
+        :param is_possible: Whether it is possible to satisfy the constraints or not
         """
         super(Constraints, self).__init__(constraint_list or [])
-        self.solver = solver
-        self.__possibility = possibility
+        self._default_timeout = 100
+        self._is_possible = is_possible
 
-    def check_possibility(self):
+    @property
+    def is_possible(self) -> bool:
         """
-
-        :return:
+        :return: True/False based on the existence of solution of constraints
         """
-        return True
+        if self._is_possible is not None:
+            return self._is_possible
+        solver = Solver()
+        solver.set_timeout(self._default_timeout)
+        for constraint in self[:]:
+            solver.add(constraint)
+        self._is_possible = solver.check() != unsat
+        return self._is_possible
 
-    def append(self, constraint):
+    def append(self, constraint: Bool) -> None:
         """
 
-        :param constraint:
+        :param constraint: The constraint to be appended
         """
         super(Constraints, self).append(constraint)
+        self._is_possible = None
 
-    def pop(self, index=-1):
+    def pop(self, index: int = -1) -> None:
         """
 
-        :param index:
+        :param index: Index to be popped from the list
         """
         raise NotImplementedError
 
-    def __copy__(self):
+    @property
+    def as_list(self) -> List[Bool]:
+        """
+        :return: returns the list of constraints
+        """
+        return self[:]
+
+    def __copy__(self) -> "Constraints":
         """
 
-        :return:
+        :return: The copied constraint List
         """
         constraint_list = super(Constraints, self).copy()
-        return Constraints(constraint_list)
+        return Constraints(constraint_list, is_possible=self._is_possible)
 
-    def __deepcopy__(self, memodict=None):
+    def __deepcopy__(self, memodict=None) -> "Constraints":
         """
 
         :param memodict:
-        :return:
+        :return: The copied constraint List
         """
         return self.__copy__()
 
-    def __add__(self, constraints):
+    def __add__(self, constraints: List[Bool]) -> "Constraints":
         """
 
         :param constraints:
-        :return:
+        :return: the new list after the + operation
         """
         constraints_list = super(Constraints, self).__add__(constraints)
-        return Constraints(constraint_list=constraints_list)
+        return Constraints(constraint_list=constraints_list, is_possible=None)
 
-    def __iadd__(self, constraints):
+    def __iadd__(self, constraints: Iterable[Bool]) -> "Constraints":
         """
 
         :param constraints:
         :return:
         """
         super(Constraints, self).__iadd__(constraints)
+        self._is_possible = None
         return self

mythril/laser/ethereum/svm.py

@@ -229,6 +229,11 @@ class LaserEVM:
             except NotImplementedError:
                 log.debug("Encountered unimplemented instruction")
                 continue
+
+            new_states = [
+                state for state in new_states if state.mstate.constraints.is_possible
+            ]
+
             self.manage_cfg(op_code, new_states)
 
             if new_states:
@@ -236,6 +241,7 @@ class LaserEVM:
             elif track_gas:
                 final_states.append(global_state)
             self.total_states += len(new_states)
+
         return final_states if track_gas else None
 
     def _add_world_state(self, global_state: GlobalState):

mythril/laser/ethereum/transaction/symbolic.py

@@ -133,7 +133,7 @@ def _setup_global_state_for_execution(laser_evm, transaction) -> None:
             )
 
         global_state.mstate.constraints += transaction.world_state.node.constraints
-        new_node.constraints = global_state.mstate.constraints
+        new_node.constraints = global_state.mstate.constraints.as_list
 
     global_state.world_state.transaction_sequence.append(transaction)
     global_state.node = new_node

tests/laser/evm_testsuite/evm_test.py

@@ -127,7 +127,8 @@ def test_vmtest(
         assert len(laser_evm.open_states) == 1
         world_state = laser_evm.open_states[0]
         model = get_model(
-            next(iter(laser_evm.nodes.values())).states[0].mstate.constraints
+            next(iter(laser_evm.nodes.values())).states[0].mstate.constraints,
+            enforce_execution_time=False,
         )
 
         for address, details in post_condition.items():