Added logging to cozy to show progress

cozy/__init__.py

@@ -11,4 +11,5 @@
 from . import terminal_state
 from . import server
 from . import types
-from . import hooks
+from . import hooks
+from . import logging

cozy/analysis.py

@@ -6,7 +6,7 @@
 
 import portion as P
 
-from . import claripy_ext
+from . import claripy_ext, logging
 from .functools_ext import *
 import collections.abc
 from .session import RunResult
@@ -361,8 +361,15 @@ def __init__(self, pre_patched: RunResult, post_patched: RunResult, ignore_addrs
             inv_addrs_pre_patched = None
             inv_addrs_post_patched = None
 
+        total_num_pairs = len(states_pre_patched) * len(states_post_patched)
+        count = 0
+
         for (i, state_pre) in enumerate(states_pre_patched):
             for (j, state_post) in enumerate(states_post_patched):
+                count += 1
+                percentage = (count / total_num_pairs) * 100.0
+                logging.info("Comparing state pair %d of %d (%.0f%% complete)", count, total_num_pairs, percentage)
+
                 if ignore_invalid_stack:
                     stack_change = state_pre.state.arch.stack_change
                     pair_ignore_addrs = ignore_addrs + [_invalid_stack_overlap(inv_addrs_pre_patched[i], inv_addrs_post_patched[j], stack_change)]

cozy/log.py

@@ -0,0 +1,46 @@
+import logging
+
+logger = logging.getLogger('cozy-log')
+def disable():
+    """
+    Disable cozy logging
+    """
+    logger.setLevel(logging.NOTSET)
+
+def set_level(level):
+    """
+    Set the level of logging
+    """
+    logger.setLevel(level)
+
+def info(*args, **kwargs):
+    """
+    Log at the info level
+    """
+    logger.info(*args, **kwargs)
+
+def warning(*args, **kwargs):
+    """
+    Log at the warning level
+    """
+    logger.warning(*args, **kwargs)
+
+def error(*args, **kwargs):
+    """
+    Log at the error level
+    """
+    logger.error(*args, **kwargs)
+
+def debug(*args, **kwargs):
+    """
+    Log at the debug level
+    """
+    logger.debug(*args, **kwargs)
+
+def critical(*args, **kwargs):
+    """
+    Log at the critical level
+    """
+    logger.critical(*args, **kwargs)
+
+set_level(logging.DEBUG)

cozy/project.py

@@ -17,14 +17,15 @@ class Project:
     :ivar dict[str | int, str] fun_prototypes: Maps function names or function addresses to their type signatures.
     """
 
-    def __init__(self, binary_path: str, fun_prototypes: dict[str | int, str] | None=None, load_debug_info: bool = False):
+    def __init__(self, binary_path: str, fun_prototypes: dict[str | int, str] | None=None, load_debug_info: bool = False, **kwargs):
         """
         Constructor for a project.
 
         :param str binary_path: The path to the binary to analyze.
         :param dict[str | int, str] | None fun_prototypes: Initial dictionary that maps function names or addresses to their type signatures. If None is passed, fun_prototypes is initialized to the empty dictionary.
+        :param kwargs: Extra arguments to pass to angr.Project
         """
-        self.angr_proj = angr.Project(binary_path, load_debug_info=load_debug_info)
+        self.angr_proj = angr.Project(binary_path, load_debug_info=load_debug_info, **kwargs)
         if fun_prototypes is None:
             self.fun_prototypes = {}
         else:

cozy/session.py

@@ -7,6 +7,7 @@
 from angr import SimStateError, SimState
 from angr.sim_manager import ErrorRecord, SimulationManager
 
+from . import logging
 from .directive import Directive, Assume, Assert, VirtualPrint, ErrorDirective, AssertType
 from .terminal_state import AssertFailedState, ErrorState, DeadendedState
 
@@ -217,6 +218,8 @@ def explore(self, simgr):
             # Explore seems to check the find list BEFORE stepping
             simgr.explore(find=find_addrs, n=1)
 
+            logging.info("Running %s: %s", self.session.proj.angr_proj.filename, str(simgr))
+
             # When using explore with a find key, angr will only put 1 state into the found stash at a time
             # This is problematic since we could create more than one error state triggered for every
             # iteration of the loop. In this case we may never iterate over all the found states before
@@ -235,11 +238,12 @@ def explore(self, simgr):
                             found_state.add_constraints(cond)
                             if not found_state.satisfiable():
                                 self.assume_warnings.append((directive, found_state))
-                                print(
-                                    "cozy WARNING: Assume for address {} was not satisfiable. In this path of execution there is no possible way for execution to proceed past this point: {}".format(
-                                        hex(directive.addr), str(cond)))
-                                if directive.info_str is not None:
-                                    print(directive.info_str)
+                                info_str = "(unknown)" if directive.info_str is None else directive.info_str
+                                logging.warning(
+                                    "Assume %s for address %s was not satisfiable. In this path of execution there is no possible way for execution to proceed past this point: %s",
+                                    info_str,
+                                    hex(directive.addr),
+                                    str(cond))
                         elif isinstance(directive, Assert):
                             cond = directive.condition_fun(found_state)
                             if directive.assert_type == AssertType.ASSERT_MUST:
@@ -312,6 +316,7 @@ class _SessionBasicExploration(_SessionExploration):
     def explore(self, simgr):
         while len(simgr.active) > 0:
             simgr.step()
+            logging.info("Running %s: %s", self.session.proj.angr_proj.filename, str(simgr))
             if self.cache_intermediate_info:
                 _save_states(simgr.active)