Fixes timeout decorator issues and radare2 parsing problems

zeratool_lib/__init__.py

@@ -0,0 +1,37 @@
+# Pickle-compatible re-exports for the host side of the gRPC round-trip.
+#
+# The docker container puts /zeratool_lib/zeratool_lib on PYTHONPATH, which
+# makes zeratool_lib.py import as a *top-level module* named `zeratool_lib`.
+# Classes defined there get __module__ == "zeratool_lib", so the pickled
+# exploit returned over gRPC references e.g. `zeratool_lib.ZeratoolExploit`.
+#
+# On the host, `zeratool_lib` is installed as a *package* (this directory).
+# We can't `from zeratool_lib.zeratool_lib import ...` here because that
+# submodule does `import formatDetector` etc. (unqualified, docker-only),
+# which would explode at import time on the host. So we redefine the small
+# pickle-relevant classes here instead. Pickle resolves classes by
+# (module, qualname) lookup and rebuilds dataclass/enum instances by field,
+# so structural equivalence is enough.
+from dataclasses import dataclass
+from enum import Enum
+
+
+class ZeratoolInputStreams(Enum):
+    """Sync names with commons.input_streams.InputStreams."""
+
+    STDIN = "STDIN"
+    ARGUMENTS = "ARG"
+
+
+@dataclass
+class ZeratoolExploit:
+    class Outcomes(Enum):
+        SHELL = "SHELL"
+        CALL_TO_WIN = "CALL_TO_WIN"
+        LEAK = "LEAK"
+
+    payload: bytes
+    outcome: Outcomes
+
+
+__all__ = ["ZeratoolExploit", "ZeratoolInputStreams"]

zeratool_lib/formatDetector.py

@@ -56,11 +56,21 @@ def checkFormat(binary_name, inputType):
     # Lame way to do a timeout
     try:
 
-        @timeout_decorator.timeout(1200)
+        # @timeout_decorator.timeout(1200)
         def exploreBinary(simgr):
             simgr.explore(find=lambda s: "type" in s.globals)
 
         exploreBinary(simgr)
+        log.info(
+            "[stashes] active=%d found=%d errored=%d deadended=%d unconstrained=%d",
+            len(simgr.active),
+            len(simgr.found),
+            len(simgr.errored),
+            len(simgr.deadended),
+            len(simgr.unconstrained),
+        )
+        for es in simgr.errored[:5]:
+            log.info("[errored] addr=%s err=%s", hex(es.state.addr), es.error)
         if "found" in simgr.stashes and len(simgr.found):
             end_state = simgr.found[0]
             run_environ["type"] = end_state.globals["type"]
@@ -70,7 +80,8 @@ def exploreBinary(simgr):
     except (KeyboardInterrupt, timeout_decorator.TimeoutError) as e:
         print("[~] Format check timed out")
 
-    if "input" in end_state.globals.keys():
+    # if "input" in end_state.globals.keys():
+    if end_state is not None and "input" in end_state.globals.keys():
         run_environ["input"] = end_state.globals["input"]
         print("[+] Triggerable with input : {}".format(end_state.globals["input"]))
 

zeratool_lib/formatExploiter.py

@@ -158,7 +158,7 @@ def rediscoverAndExploit(binary_name, properties, stack_position, leak_format):
     # Lame way to do a timeout
     try:
 
-        @timeout_decorator.timeout(1200)
+        # @timeout_decorator.timeout(1200)
         def exploreBinary(simgr):
             simgr.explore(find=lambda s: "type" in s.globals)
 
@@ -173,15 +173,18 @@ def exploreBinary(simgr):
         log.info("[~] Format check timed out")
 
     if inputType == "STDIN" and end_state is not None:
-        stdin_str = str(end_state.posix.dumps(0))
-        log.info("[+] Triggerable with STDIN : {}".format(stdin_str))
+        # `posix.dumps(0)` already returns bytes. The previous str(...) wrap
+        # produced a Python repr literal ("b'...'") which broke downstream
+        # consumers expecting bytes (ZeratoolExploit.payload, hexdump, etc.).
+        stdin_bytes = end_state.posix.dumps(0)
+        log.info("[+] Triggerable with STDIN : {!r}".format(stdin_bytes))
 
-        return stdin_str, end_state.globals["fmt_outcome"]
+        return stdin_bytes, end_state.globals["fmt_outcome"]
     elif inputType == "ARG" and end_state is not None:
-        arg_str = str(end_state.solver.eval(arg, cast_to=str))
-        run_environ["input"] = arg_str
+        arg_bytes = end_state.solver.eval(arg, cast_to=bytes)
+        run_environ["input"] = arg_bytes
 
-        return arg_str, end_state.globals["fmt_outcome"]
+        return arg_bytes, end_state.globals["fmt_outcome"]
 
 
 def get_num_constraints(chop_byte, state):
@@ -190,7 +193,7 @@ def get_num_constraints(chop_byte, state):
     # Do any constraints mention this BV?
     for constraint in constraints:
         if any(
-            chop_byte.structurally_match(x) for x in constraint.recursive_children_asts
+            chop_byte.structurally_match(x) for x in constraint.children_asts()
         ):
             i += 1
     # log.info("{} : {} : {}".format(chop_byte,i,state.solver.eval(chop_byte,cast_to=bytes)))
@@ -264,10 +267,19 @@ def checkExploitable(self, fmt):
 
         var_loc = solv(printf_arg)
 
+        if printf_model.format_string_ptr_is_read_only(state.project, var_loc):
+            log.debug(
+                "Skipping format exploit hook for read-only format pointer %s",
+                hex(var_loc),
+            )
+            return False
+
         # Parts of this argument could be symbolic, so we need
         # to check every byte
         var_data = state.memory.load(var_loc, max_read_len)
         var_len = get_max_strlen(state, var_data)
+        if var_len == 0:
+            var_len = min(max_read_len, 256)
 
         fmt_len = self._sim_strlen(fmt)
         # if len(state.solver.eval_upto(fmt_len,2)) > 1:
@@ -320,8 +332,19 @@ def checkExploitable(self, fmt):
             state_copy = None
             results_n = None
 
+            # Snapshot the pre-loop state once. Each GOT iteration must start
+            # from this; otherwise constrainBytes() keeps stacking constraints
+            # on `state` (which aliases self.state) so the input bytes get
+            # pinned to iter-0's address (e.g. __gmon_start__) and every later
+            # solver.eval(user_input) returns that stale payload regardless of
+            # which GOT entry we now intend to overwrite.
+            original_state = state.copy()
+
             for got_name, got_addr in list(properties["protections"]["got"].items()):
                 #                for got_name,got_addr in [(x,y) for (x,y) in properties['protections']['got'].items() if x in " exit"]: #debug for hard_format
+                # Reset state to the pre-loop snapshot so this iteration's
+                # constrainBytes only sees its own constraints.
+                state = original_state.copy()
                 backup_state = state.copy()
                 log.info("[+] Overwiting {} at {}".format(got_name, hex(got_addr)))
 
@@ -377,17 +400,34 @@ def checkExploitable(self, fmt):
                 results = sendExploit(binary_name, properties, user_input, leak_format)
                 exploit_results = {}
                 if results["success"] == True:
-                    exploit_results["success"] = results["success"]
-                    exploit_results["input"] = user_input
+                    # Promote the working payload onto self.state and bail.
+                    # Without this, the simple-success path used to fall
+                    # through into the (state_copy-gated) verification block,
+                    # which is skipped on the very first successful iteration
+                    # (state_copy is still None) and we'd never return True.
+                    var_value = self.state.memory.load(var_loc, var_value_length)
+                    self.constrainBytes(
+                        self.state,
+                        var_value,
+                        var_loc,
+                        position,
+                        var_value_length,
+                        strVal=format_payload,
+                    )
+                    log.info("[+] Vulnerable path found {}".format(repr(user_input)))
+                    self.state.globals["type"] = "Format"
+                    self.state.globals["position"] = position
+                    self.state.globals["length"] = greatest_count
                     self.state.globals["fmt_outcome"] = results["fmt_outcome"]
+                    return True
                 else:  # Maybe angr still messed up the pointer
                     log.info("[-] Payload launch failed. Fixing angr stack pointer")
 
                     # Find the last basic block executed
 
                     first_input = state.posix.dumps(0)
 
-                    end_eip = state.se.eval(state.regs.pc)
+                    end_eip = state.solver.eval(state.regs.pc)
 
                     last_bb = [
                         x
@@ -543,8 +583,10 @@ def constrainBytes(self, state, symVar, loc, position, length, strVal="%x_"):
         for i in range(length):
             strValIndex = i % len(strVal)
             curr_byte = self.state.memory.load(loc + i, 1).get_byte(0)
-            constraint = state.se.And(strVal[strValIndex] == curr_byte)
-            if state.se.satisfiable(extra_constraints=[constraint]):
+            # `state.solver` doesn't expose `.And`; wrapping a single boolean
+            # AST in And(...) is a no-op so we just use the comparison directly.
+            constraint = strVal[strValIndex] == curr_byte
+            if state.solver.satisfiable(extra_constraints=[constraint]):
                 state.add_constraints(constraint)
             else:
                 log.info(
@@ -553,7 +595,10 @@ def constrainBytes(self, state, symVar, loc, position, length, strVal="%x_"):
                     )
                 )
 
-    def run(self, _, fmt):
+    def run(self, fmt):
+        # See note in printf_model.printFormat.run: angr's printf takes a
+        # single fmt arg; the upstream `(self, _, fmt)` signature is for
+        # __printf_chk and was forwarding stack garbage to super().run.
         if not self.checkExploitable(fmt):
             return super(type(self), self).run(fmt)
 
@@ -646,7 +691,10 @@ def sendExploit(binary_name, properties, input_string, leak_format):
 
     log.info(results)
     send_results["success"] = False
-    if not hadIssue and re.match(leak_format, results):
+    # Same str/bytes guard as formatLeak.checkLeak: no leak target -> no match.
+    if isinstance(leak_format, str):
+        leak_format = leak_format.encode() if leak_format else b""
+    if not hadIssue and leak_format and re.match(leak_format, results):
         send_results["success"] = True
         send_results["payload"] = input_string
         send_results["fmt_outcome"] = "leak"

zeratool_lib/formatLeak.py

@@ -6,6 +6,13 @@
 
 
 def checkLeak(binary_name, properties, leak_format) -> bytes:
+    # No leak target requested (e.g. plain shellcode exploit) -> nothing to do.
+    # Avoids `re.match("", bytes)` TypeError when the default `""` falls through.
+    if not leak_format:
+        return None
+    if isinstance(leak_format, str):
+        leak_format = leak_format.encode()
+
     full_string = b""
     run_count = 50
 

zeratool_lib/overflowDetector.py

@@ -49,7 +49,7 @@ def checkOverflow(binary_name, inputType):
     # Lame way to do a timeout
     try:
 
-        @timeout_decorator.timeout(120)
+        # @timeout_decorator.timeout(120)
         def exploreBinary(simgr):
             simgr.explore(
                 find=lambda s: "type" in s.globals, step_func=overflow_detect_filter

zeratool_lib/overflowExploiter.py

@@ -146,7 +146,7 @@ def exploitOverflow(binary_name, properties, inputType):
     simgr.explore(find=lambda s: "type" in s.globals, step_func=step_func)
     try:
 
-        @timeout_decorator.timeout(1200)
+        # @timeout_decorator.timeout(1200)
         def exploreBinary(simgr):
             simgr.explore(find=lambda s: "type" in s.globals, step_func=step_func)
 

zeratool_lib/overflowRemoteLeaker.py

@@ -69,7 +69,7 @@ def leak_remote_functions(binary_name, properties, inputType):
     # Lame way to do a timeout
     try:
 
-        @timeout_decorator.timeout(1200)
+        # @timeout_decorator.timeout(1200)
         def exploreBinary(simgr):
             simgr.explore(
                 find=lambda s: "libc" in s.globals, step_func=leak_remote_libc_functions