Fix symbolic syscall arg concretization

manticore/native/cpu/abstractcpu.py

@@ -89,7 +89,9 @@ class ConcretizeRegister(CpuException):
     Raised when a symbolic register needs to be concretized.
     """
 
-    def __init__(self, cpu, reg_name, message=None, policy="MINMAX"):
+    def __init__(
+        self, cpu: "Cpu", reg_name: str, message: Optional[str] = None, policy: str = "MINMAX",
+    ):
         self.message = message if message else f"Concretizing {reg_name}"
 
         self.cpu = cpu

manticore/native/memory.py

@@ -11,6 +11,7 @@
     BitVecConstant,
     expression,
     issymbolic,
+    Expression,
 )
 from ..native.mappings import mmap, munmap
 from ..utils.helpers import interval_intersection
@@ -19,7 +20,7 @@
 import functools
 import logging
 
-from typing import Dict, Generator, Iterable, List, MutableMapping, Optional, Set
+from typing import Dict, Generator, Iterable, List, MutableMapping, Optional, Set, Union
 
 logger = logging.getLogger(__name__)
 
@@ -58,7 +59,14 @@ class ConcretizeMemory(MemoryException):
     Raised when a symbolic memory cell needs to be concretized.
     """
 
-    def __init__(self, mem, address, size, message=None, policy="MINMAX"):
+    def __init__(
+        self,
+        mem: "Memory",
+        address: Union[int, Expression],
+        size: int,
+        message: Optional[str] = None,
+        policy: str = "MINMAX",
+    ):
         if message is None:
             self.message = f"Concretizing memory address {address} size {size}"
         else:

manticore/native/state.py

@@ -1,7 +1,15 @@
+from collections import namedtuple
+from typing import Any, NamedTuple
+
 from ..core.state import StateBase, Concretize, TerminateState
 from ..native.memory import ConcretizeMemory, MemoryException
 
 
+class CheckpointData(NamedTuple):
+    pc: Any
+    last_pc: Any
+
+
 class State(StateBase):
     @property
     def cpu(self):
@@ -17,6 +25,14 @@ def mem(self):
         """
         return self._platform.current.memory
 
+    def _rollback(self, checkpoint_data: CheckpointData) -> None:
+        """
+        Rollback state to previous values in checkpoint_data
+        """
+        # Keep in this form to make sure we don't miss restoring any newly added
+        # data. Make sure the order is correct
+        self.cpu.PC, self.cpu._last_pc = checkpoint_data
+
     def execute(self):
         """
         Perform a single step on the current state
@@ -25,28 +41,36 @@ def execute(self):
             ConcretizeRegister,
         )  # must be here, otherwise we get circular imports
 
+        checkpoint_data = CheckpointData(pc=self.cpu.PC, last_pc=self.cpu._last_pc)
         try:
             result = self._platform.execute()
 
         # Instead of State importing SymbolicRegisterException and SymbolicMemoryException
         # from cpu/memory shouldn't we import Concretize from linux, cpu, memory ??
         # We are forcing State to have abstractcpu
-        except ConcretizeRegister as e:
+        except ConcretizeRegister as exc:
+            # Need to define local variable to use in closure
+            e = exc
             expression = self.cpu.read_register(e.reg_name)
 
-            def setstate(state, value):
-                state.cpu.write_register(setstate.e.reg_name, value)
+            def setstate(state: State, value):
+                state.cpu.write_register(e.reg_name, value)
 
-            setstate.e = e
+            self._rollback(checkpoint_data)
             raise Concretize(str(e), expression=expression, setstate=setstate, policy=e.policy)
-        except ConcretizeMemory as e:
+        except ConcretizeMemory as exc:
+            # Need to define local variable to use in closure
+            e = exc
             expression = self.cpu.read_int(e.address, e.size)
 
-            def setstate(state, value):
-                state.cpu.write_int(setstate.e.address, value, setstate.e.size)
+            def setstate(state: State, value):
+                state.cpu.write_int(e.address, value, e.size)
 
-            setstate.e = e
+            self._rollback(checkpoint_data)
             raise Concretize(str(e), expression=expression, setstate=setstate, policy=e.policy)
+        except Concretize as e:
+            self._rollback(checkpoint_data)
+            raise e
         except MemoryException as e:
             raise TerminateState(str(e), testcase=True)
 

manticore/platforms/platform.py

@@ -43,14 +43,6 @@ def __init__(self, idx, name):
         super().__init__(msg)
 
 
-class ConcretizeSyscallArgument(OSException):
-    def __init__(self, reg_num, message="Concretizing syscall argument", policy="SAMPLED"):
-        self.reg_num = reg_num
-        self.message = message
-        self.policy = policy
-        super().__init__(message)
-
-
 class Platform(Eventful):
     """
     Base class for all platforms e.g. operating systems or virtual machines.

tests/native/binaries/symbolic_read_count.c

@@ -0,0 +1,26 @@
+// Compiled on Ubuntu 18.04 Manticore Docker image with
+//   gcc -static symbolic_read_count.c -o symbolic_read_count
+
+#include <stdio.h>
+#include <stdlib.h>
+
+int main(int argc, char **argv) {
+  // Need at least one argument
+  if (argc != 2) {
+    return -1;
+  }
+
+  // Just get the char ordinal value
+  unsigned int count = argv[1][0];
+  if (count > 9) {
+    return 0;
+  }
+
+  // Yes... this is very unsafe
+  char *buf[10];
+  int sz = read(0, buf, count);
+  if (sz > 0) {
+    printf("WIN: Read more than zero data\n");
+  }
+  return sz;
+}

tests/native/test_syscalls.py

@@ -9,12 +9,36 @@
 import os
 import errno
 import re
+from glob import glob
+
+from manticore.native import Manticore
 
 from manticore.platforms import linux, linux_syscall_stubs
 from manticore.platforms.linux import SymbolicSocket
 from manticore.platforms.platform import SyscallNotImplemented, logger as platform_logger
 
 
+def test_symbolic_syscall_arg() -> None:
+    BIN_PATH = os.path.join(os.path.dirname(__file__), "binaries", "symbolic_read_count")
+    tmp_dir = tempfile.TemporaryDirectory(prefix="mcore_test_")
+    m = Manticore(BIN_PATH, argv=["+"], workspace_url=str(tmp_dir.name))
+
+    m.run()
+    m.finalize()
+
+    found_win_msg = False
+    win_msg = "WIN: Read more than zero data"
+    outs_glob = f"{str(m.workspace)}/test_*.stdout"
+    # Search all output messages
+    for output_p in glob(outs_glob):
+        with open(output_p) as f:
+            if win_msg in f.read():
+                found_win_msg = True
+                break
+
+    assert found_win_msg, f'Did not find win message in {outs_glob}: "{win_msg}"'
+
+
 class LinuxTest(unittest.TestCase):
     _multiprocess_can_split_ = True
     BIN_PATH = os.path.join(os.path.dirname(__file__), "binaries", "basic_linux_amd64")