Fix bugs and issues found with static type checking (#307)

* Fix issues found with static type checking
* Cleanup for CI, finalize for majority of Python code
* Update CI configuration for static checking

Author: Alan

.github/workflows/ci.yml

@@ -20,11 +20,12 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install pyflakes
+          pip install pyflakes mypy
       - name: Lint
         run: |
           pyflakes bin/deepstate/*.py
           pyflakes tests/*.py
+          mypy --strict-optional --config-file mypy.ini bin/deepstate
   build:
     strategy:
       matrix:

bin/deepstate/core/base.py

@@ -20,7 +20,7 @@
 import argparse
 import configparser
 
-from typing import Dict, ClassVar, Optional, Union, List
+from typing import Dict, ClassVar, Optional, Union, List, Any
 
 
 L = logging.getLogger("deepstate.core.base")
@@ -41,7 +41,7 @@ class AnalysisBackend(object):
   COMPILER: ClassVar[Optional[str]] = None
 
   # temporary attribute for argparsing, and should be used to build up object attributes
-  _ARGS: ClassVar[Optional[Dict[str, str]]] = None
+  _ARGS: ClassVar[Optional[argparse.Namespace]] = None
 
   # temporary attribute for parser instantiation, should be used to check if user parsed args
   parser: ClassVar[Optional[argparse.ArgumentParser]] = None
@@ -52,20 +52,30 @@ def __init__(self):
 
 
   @classmethod
-  def parse_args(cls) -> None:
+  def parse_args(cls) -> Optional[argparse.Namespace]:
     """
     Base root-level argument parser. After the executors initializes its application-specific arguments, and the frontend
     builds up further with analysis-specific arguments, this base parse_args finalizes with all other required args every
     executor should consume.
     """
-    parser = cls.parser
+
+    if cls._ARGS:
+      L.debug("Returning already-parsed arguments")
+      return cls._ARGS
+
+    # checks if frontend executor already implements an argparser, since we want to extend on that.
+    if cls.parser is not None:
+      parser: argparse.ArgumentParser = cls.parser
+    else:
+      parser = argparse.ArgumentParser(description="Use {} as a backend for DeepState".format(cls.NAME))
 
     # Compilation/instrumentation support, only if COMPILER is set
     # TODO: extends compilation interface for symex engines that "compile" source to
     # binary, IR format, or boolean expressions for symbolic VM to reason with
     if cls.COMPILER:
       L.debug("Adding compilation support since a compiler was specified")
 
+      # type: ignore
       compile_group = parser.add_argument_group("Compilation and Instrumentation")
       compile_group.add_argument("--compile_test", type=str,
         help="Path to DeepState test source for compilation and instrumentation by analysis tool.")
@@ -86,8 +96,8 @@ def parse_args(cls) -> None:
 
     # Analysis-related configurations
     parser.add_argument(
-      "-o", "--output_test_dir", type=str, default="{}_out".format(cls.NAME),
-      help="Output directory where tests will be saved (default is `{FUZZER}_out`).")
+      "-o", "--output_test_dir", type=str, default="out",
+      help="Output directory where tests will be saved (default is `out`).")
 
     parser.add_argument(
       "-c", "--config", type=str,
@@ -113,25 +123,27 @@ def parse_args(cls) -> None:
       help="Other DeepState flags to pass to harness before execution, in format `--arg=val`.")
 
     args = parser.parse_args()
+
+    # from parsed arguments, modify dict copy if configuration is specified
     _args: Dict[str, str] = vars(args)
 
     # if configuration is specified, parse and replace argument instantiations
     if args.config:
-      _args.update(cls.build_from_config(args.config))
+      _args.update(cls.build_from_config(args.config)) # type: ignore
 
       # Cleanup: force --no_exit_compile to be on, meaning if user specifies a `[test]` section,
       # execution will continue. Delete config as well
-      _args["no_exit_compile"] = True
+      _args["no_exit_compile"] = True # type: ignore
       del _args["config"]
 
     cls._ARGS = args
     return cls._ARGS
 
 
-  ConfigType = Dict[str, Dict[str, Union[str, List[str]]]]
+  ConfigType = Dict[str, Dict[str, Any]]
 
   @staticmethod
-  def build_from_config(config: str, allowed_keys: Optional[List[str]] = None,  include_sections: bool = False) -> Union[ConfigType, Dict[str, str]]:
+  def build_from_config(config: str, allowed_keys: Optional[List[str]] = None, include_sections: bool = False) -> Union[ConfigType, Dict[str, Any]]:
     """
     Simple auxiliary helper that does safe and correct parsing of DeepState configurations. This can be used
     in the following manners:
@@ -145,7 +157,7 @@ def build_from_config(config: str, allowed_keys: Optional[List[str]] = None,  in
     :param include_sections: if true, parse all sections, and return a ConfigType where keys are section names
     """
 
-    context: ConfigType = dict()
+    context: ConfigType = dict() # type: ignore
 
     # reserved sections are ignored by executors, but can be used by other auxiliary tools
     # to reason about with.
@@ -164,7 +176,7 @@ def build_from_config(config: str, allowed_keys: Optional[List[str]] = None,  in
     parser = configparser.SafeConfigParser()
     parser.read(config)
 
-    for section, kv in parser._sections.items():
+    for section, kv in parser._sections.items(): # type: ignore
 
       # if `include_sections` is not set, parse only from allowed_sections
       if not include_sections:
@@ -191,7 +203,7 @@ def build_from_config(config: str, allowed_keys: Optional[List[str]] = None,  in
         else:
           _context[key] = val
 
-    return context
+    return context # type: ignore
 
 
   def init_from_dict(self, _args: Optional[Dict[str, str]] = None) -> None:

bin/deepstate/core/fuzz.py

@@ -82,13 +82,17 @@ def __init__(self, envvar: str = "PATH") -> None:
       compiler_paths: List[str] = [f"{path}/{compiler}" for path in potential_paths if os.path.isfile(path + '/' + compiler)]
       if len(compiler_paths) == 0:
 
-        # if not in envvar, check to see if user supplied absolute path
+        # if not in earlier envvar, check to see if user supplied absolute path
         if os.path.isfile(compiler):
           self.compiler: str = compiler
 
         # .. or check if in $PATH before tossing exception
         else:
-          for path in os.environ.get("PATH").split(os.pathsep):
+          path_env: Optional[str] = os.environ.get("PATH")
+          if path_env is None:
+            raise FuzzFrontendError("$PATH envvar is not defined.")
+
+          for path in path_env.split(os.pathsep):
             compiler_path: str = os.path.join(path, compiler)
 
             L.debug(f"Checking if `{compiler_path}` is a valid compiler path")
@@ -106,6 +110,9 @@ def __init__(self, envvar: str = "PATH") -> None:
 
       L.debug(f"Initialized compiler: {self.compiler}")
 
+    # store envvar for re-use across API
+    self.env: str = env
+
     # in case name supplied as `bin/fuzzer`, strip executable name
     self.name: str = fuzzer_name.split('/')[-1] if '/' in fuzzer_name else fuzzer_name
     L.debug(f"Fuzzer name: {self.name}")
@@ -125,7 +132,7 @@ def __init__(self, envvar: str = "PATH") -> None:
     self.num_workers: int = 1
 
     self.compile_test: Optional[str] = None
-    self.compiler_args: List[str] = []
+    self.compiler_args: Optional[str] = None
     self.out_test_name: str = "out"
 
     self.enable_sync: bool = False
@@ -134,6 +141,7 @@ def __init__(self, envvar: str = "PATH") -> None:
     self.sync_dir: str = "out_sync"
 
     self.which_test: Optional[str] = None
+    self.post_stats: bool = False
 
 
   def __repr__(self) -> str:
@@ -191,6 +199,12 @@ def parse_args(cls) -> Optional[argparse.Namespace]:
       help="Time in seconds the executor should sync to sync directory (default is 5 seconds).")
 
 
+    # Post-processing
+    post_group = parser.add_argument_group("Execution Post-processing")
+    post_group.add_argument("--post_stats", action="store_true",
+      help="Output post-fuzzing statistics to user (if any).")
+
+
     # Miscellaneous options
     parser.add_argument(
       "--fuzzer_help", action="store_true",
@@ -201,6 +215,8 @@ def parse_args(cls) -> Optional[argparse.Namespace]:
     cls.parser = parser
     super(FuzzerFrontend, cls).parse_args()
 
+    return None
+
 
   def print_help(self) -> None:
     """
@@ -214,18 +230,18 @@ def print_help(self) -> None:
   ##############################################
 
 
-  def compile(self, lib_path: str, flags: List[str], _out_bin: str, env = os.environ.copy()) -> Optional[str]:
+  def compile(self, lib_path: str, flags: List[str], _out_bin: str, env = os.environ.copy()) -> None:
     """
     Provides a simple interface that allows the user to compile a test harness
     with instrumentation using the specified compiler. Users should implement an
     inherited method that constructs the arguments necessary, and then pass it to the
-    base object. Returns string of generated binary if successful.
+    base object.
 
     `compile()` also supports compiling arbitrary harnesses without instrumentation if a compiler
     isn't set.
 
     :param lib_path: path to DeepState static library for linking
-    :param flags: list of compiler flags (TODO: parse from compilation database)
+    :param flags: list of compiler flags (TODO: support parsing from compilation database path)
     :param _out_bin: name of linked test harness binary
     :param env: optional envvars to set during compilation
     """
@@ -237,7 +253,7 @@ def compile(self, lib_path: str, flags: List[str], _out_bin: str, env = os.envir
       raise FuzzFrontendError(f"User-specified test binary conflicts with compiling from source.")
 
     if not os.path.isfile(lib_path):
-      raise FuzzFrontendError("No {}-instrumented DeepState static library found in {}".format(cls, lib_path))
+      raise FuzzFrontendError("No {}-instrumented DeepState static library found in {}".format(self, lib_path))
     L.debug(f"Static library path: {lib_path}")
 
     # initialize compiler envvars
@@ -246,8 +262,7 @@ def compile(self, lib_path: str, flags: List[str], _out_bin: str, env = os.envir
     L.debug(f"CC={env['CC']} and CXX={env['CXX']}")
 
     # initialize command with prepended compiler
-    compiler_args = ["-std=c++11", self.compile_test] + flags + \
-                    ["-o", _out_bin]
+    compiler_args: List[str] = ["-std=c++11", self.compile_test] + flags + ["-o", _out_bin] # type: ignore
     compile_cmd = [self.compiler] + compiler_args
     L.debug(f"Compilation command: {str(compile_cmd)}")
 
@@ -259,7 +274,7 @@ def compile(self, lib_path: str, flags: List[str], _out_bin: str, env = os.envir
       raise FuzzFrontendError(f"{self.compiler} interrupted due to exception:", e)
 
     # extra check if target binary was successfully compiled, and set that as target binary
-    out_bin = os.path.join(os.environ.get("PWD"), _out_bin)
+    out_bin = os.path.join(os.getcwd(), _out_bin)
     if os.path.exists(out_bin):
       self.binary = out_bin
 
@@ -351,7 +366,7 @@ def _dict_to_cmd(cmd_dict: Dict[str, Optional[str]]) -> List[Optional[str]]:
     return cmd_args
 
 
-  def build_cmd(self, cmd_dict: Dict[str, Optional[str]], input_symbol: str = "@@") -> Dict[str, Optional[str]]:
+  def build_cmd(self, cmd_dict: Dict[str, Any], input_symbol: str = "@@") -> Dict[str, Optional[str]]:
     """
     Helper method to be invoked by child fuzzer class's cmd() property method in order
     to finalize command called by the fuzzer executable with appropriate arguments for the
@@ -409,15 +424,15 @@ def run(self, compiler: Optional[str] = None, no_exec: bool = False):
       self.pre_exec()
 
     # initialize cmd from property
-    command = [self.fuzzer] + self._dict_to_cmd(self.cmd)
+    command = [self.fuzzer] + self._dict_to_cmd(self.cmd) # type: ignore
 
     # prepend compiler that invokes fuzzer
     if compiler:
       command.insert(0, compiler)
 
     results: List[int] = []
     pool = multiprocessing.Pool(processes=self.num_workers)
-    results = pool.apply_async(self._run, args=(command,))
+    results = pool.apply_async(self._run, args=(command,)) # type: ignore
 
     pool.close()
     pool.join()

bin/deepstate/core/symex.py

@@ -14,27 +14,18 @@
 
 import logging
 logging.basicConfig()
-logging.addLevelName(15, "TRACE")
 
+import os
+import struct
 import argparse
-#import md5
 import hashlib
 import functools
-import os
-import struct
 
 from deepstate.core.base import AnalysisBackend
 
 
-class TestInfo(object):
-  """Represents a `DeepState_TestInfo` data structure from the program, as
-  well as associated meta-data about the test."""
-  def __init__(self, ea, name, file_name, line_number):
-    self.ea = ea
-    self.name = name
-    self.file_name = file_name
-    self.line_number = line_number
-
+LOGGER = logging.getLogger("deepstate")
+LOGGER.setLevel(os.environ.get("DEEPSTATE_LOG", "INFO").upper())
 
 LOG_LEVEL_DEBUG = 0
 LOG_LEVEL_TRACE = 1
@@ -44,23 +35,29 @@ def __init__(self, ea, name, file_name, line_number):
 LOG_LEVEL_EXTERNAL = 5
 LOG_LEVEL_FATAL = 6
 
-
-LOGGER = logging.getLogger("deepstate")
-LOGGER.setLevel(logging.DEBUG)
-
-LOGGER.trace = functools.partial(LOGGER.log, 15)
-logging.TRACE = 15
+LOGGER.trace = functools.partial(LOGGER.log, 15) # type: ignore
+logging.TRACE = 15 # type: ignore
 
 LOG_LEVEL_TO_LOGGER = {
   LOG_LEVEL_DEBUG: LOGGER.debug,
-  LOG_LEVEL_TRACE: LOGGER.trace,
+  LOG_LEVEL_TRACE: LOGGER.trace, # type: ignore
   LOG_LEVEL_INFO: LOGGER.info,
   LOG_LEVEL_WARNING: LOGGER.warning,
   LOG_LEVEL_ERROR: LOGGER.error,
   LOG_LEVEL_FATAL: LOGGER.critical
 }
 
 
+class TestInfo(object):
+  """Represents a `DeepState_TestInfo` data structure from the program, as
+  well as associated meta-data about the test."""
+  def __init__(self, ea, name, file_name, line_number):
+    self.ea = ea
+    self.name = name
+    self.file_name = file_name
+    self.line_number = line_number
+
+
 class Stream(object):
   def __init__(self, entries):
     self.entries = entries
@@ -146,13 +143,13 @@ def parse_args(cls):
 
   @property
   def context(self):
-    """Gives convenient property-based access to a dictionary holding state-
-    local varaibles."""
+    """Gives convenient property-based access to a dictionary holding state-local variables."""
     return self.get_context()
 
+
   def read_c_string(self, ea, concretize=True, constrain=False):
-    """Read a NUL-terminated string from `ea`."""
-    assert isinstance(ea, (int))
+    """Read a NULL-terminated string from address `ea`."""
+
     chars = []
     while True:
       b, ea = self.read_uint8_t(ea, concretize=concretize, constrain=constrain)
@@ -173,6 +170,7 @@ def read_c_string(self, ea, concretize=True, constrain=False):
     else:
       return chars, next_ea
 
+
   def _read_test_info(self, ea):
     """Read in a `DeepState_TestInfo` info structure from memory."""
     prev_test_ea, ea = self.read_uintptr_t(ea)