Fixing style errors.

Author: stickler-ci

boofuzz/cli.py

@@ -42,9 +42,13 @@ def cli():
 )
 @click.option("--procmon-host", help="Process monitor port host or IP")
 @click.option("--procmon-port", type=int, default=DEFAULT_PROCMON_PORT, help="Process monitor port")
-@click.option("--stdout", type=click.Choice(["HIDE", "CAPTURE", "MIRROR"], case_sensitive=False), default="MIRROR",
-              help="How to handle stdout (and stderr) of target. CAPTURE saves output for crash reporting but can "
-                   "slow down fuzzing.")
+@click.option(
+    "--stdout",
+    type=click.Choice(["HIDE", "CAPTURE", "MIRROR"], case_sensitive=False),
+    default="MIRROR",
+    help="How to handle stdout (and stderr) of target. CAPTURE saves output for crash reporting but can "
+    "slow down fuzzing.",
+)
 @click.option("--tui/--no-tui", help="Enable TUI")
 @click.option("--text-dump/--no-text-dump", help="Enable full text dump of logs", default=False)
 @click.option("--feature-check", is_flag=True, help="Run a feature check instead of a fuzz test", default=False)
@@ -64,12 +68,18 @@ def cli():
     type=int,
     help="Record this many cases before each failure. Set to 0 to record all test cases (high disk space usage!).",
 )
-@click.option("--qemu/--no-qemu", is_flag=True, default=False,
-              help="Experimental: Enable QEMU mode with code coverage feedback; requires afl-qemu-trace")
+@click.option(
+    "--qemu/--no-qemu",
+    is_flag=True,
+    default=False,
+    help="Experimental: Enable QEMU mode with code coverage feedback; requires afl-qemu-trace",
+)
 @click.option("--qemu-path", help="afl-qemu-trace path; looks in PATH by default")
 @click.option("--web-port", type=int, default=constants.DEFAULT_WEB_UI_PORT, help="port for web GUI")
 @click.option("--restart-interval", type=int, help="restart every n test cases")
-@click.option("--target-start-wait", type=float, default=0, help="wait n seconds for target to settle in before fuzzing")
+@click.option(
+    "--target-start-wait", type=float, default=0, help="wait n seconds for target to settle in before fuzzing"
+)
 @click.pass_context
 def fuzz(
     ctx,

boofuzz/helpers.py

@@ -29,8 +29,9 @@
 COLOR_PAIR_MAGENTA = 6
 COLOR_PAIR_BLACK = 7
 
-sigmap = dict((k, v) for v, k in reversed(sorted(signal.__dict__.items()))
-              if v.startswith('SIG') and not v.startswith('SIG_'))
+sigmap = dict(
+    (k, v) for v, k in reversed(sorted(signal.__dict__.items())) if v.startswith("SIG") and not v.startswith("SIG_")
+)
 
 test_step_info = {
     "test_case": {
@@ -310,7 +311,7 @@ def udp_checksum(msg, src_addr, dst_addr):
     # If the packet is too big, the checksum is undefined since len(msg)
     # won't fit into two bytes. So we just pick our best definition.
     # "Truncate" the message as it appears in the checksum.
-    msg = msg[0: ip_constants.UDP_MAX_LENGTH_THEORETICAL]
+    msg = msg[0 : ip_constants.UDP_MAX_LENGTH_THEORETICAL]
 
     return ipv4_checksum(_udp_checksum_pseudo_header(src_addr, dst_addr, len(msg)) + msg)
 

boofuzz/sessions.py

@@ -1013,7 +1013,9 @@ def _stop_targets(self):
                 target.vmcontrol.restart_target()
             else:
                 for monitor in target.monitors:
-                    self._fuzz_data_logger.log_info("Stopping target process using {}".format(monitor.__class__.__name__))
+                    self._fuzz_data_logger.log_info(
+                        "Stopping target process using {}".format(monitor.__class__.__name__)
+                    )
                     monitor.stop_target()
 
     def _restart_target(self, target):
@@ -1533,7 +1535,8 @@ def _generate_n_mutations_for_path(self, path, depth, base_mutations=None):
         if base_mutations is not None:
             skip_elements.update(m.qualified_name for m in base_mutations)
         for mutations in self._generate_n_mutations_for_path_recursive(
-                path, depth=depth, skip_elements=skip_elements, base_mutations=base_mutations):
+            path, depth=depth, skip_elements=skip_elements, base_mutations=base_mutations
+        ):
             if not self._mutations_contain_duplicate(mutations):
                 self.total_mutant_index += 1
                 yield MutationContext(message_path=path, mutations={n.qualified_name: n for n in mutations})
@@ -1662,13 +1665,13 @@ def _parse_mutation_names(self, mutation_names):
             try:
                 request = self.nodes[request_name]
             except KeyError:
-                raise Exception(
-                    "Request {0} not found in blocks.REQUESTS: {1}".format(request_name, self.nodes))
+                raise Exception("Request {0} not found in blocks.REQUESTS: {1}".format(request_name, self.nodes))
             try:
                 fuzzable = request.names[qualified_name]
             except KeyError:
-                raise Exception("Name {0} not found in request {1}.names: {2}".format(
-                    qualified_name, request_name, request.names))
+                raise Exception(
+                    "Name {0} not found in request {1}.names: {2}".format(qualified_name, request_name, request.names)
+                )
             mutations += next(itertools.islice(fuzzable.get_mutations(), index, index + 1))
         return mutations
 

boofuzz/utils/debugger_thread_qemu.py

@@ -35,6 +35,7 @@
 
 class ForkServer:
     """Implements the AFL fork server protocol. Used by DebuggerThreadQemu."""
+
     def __init__(self, args, hide_output):
         self.hide_output = hide_output
         self.pid = None
@@ -53,8 +54,7 @@ def __init__(self, args, hide_output):
             self.parent()
 
     def child(self, args):
-        """Execute afl-qemu-trace with appropriate inputs: target command args, env var settings, and file descriptors.
-        """
+        """Execute afl-qemu-trace with appropriate inputs: target command args, env var settings, and file descriptors."""
         os.dup2(self.forkserv_fd_to_server_out, AFL_FORKSRV_FD)
         os.dup2(self.forkserv_fd_from_server_in, AFL_FORKSRV_FD + 1)
 
@@ -68,9 +68,10 @@ def child(self, args):
         os.close(self.forkserv_fd_to_server_out)
         os.close(self.forkserv_fd_from_server_in)
         os.close(self.forkserv_fd_from_server_out)
-        env = {"QEMU_LOG": "nochain",
-               AFL_SHM_ENV_VAR: str(self.shm_id),
-               }
+        env = {
+            "QEMU_LOG": "nochain",
+            AFL_SHM_ENV_VAR: str(self.shm_id),
+        }
         os.execve(QEMU_PATH, ["afl-qemu-trace"] + args, env)
 
     def parent(self):
@@ -79,8 +80,7 @@ def parent(self):
         os.read(self.forkserv_fd_from_server_out, 4)
 
     def run(self):  # only the parent runs run()
-        """Runs the testcase in QEMU (by sending a command to the fork server) and returns the pid.
-        """
+        """Runs the testcase in QEMU (by sending a command to the fork server) and returns the pid."""
         os.write(self.forkserv_fd_to_server_in, b"\0\0\0\0")  # Tell AFL Fork Server to start the target
         pid = struct.unpack("I", os.read(self.forkserv_fd_from_server_out, 4))[0]  # Read PID from Fork Server
         self.pid = pid
@@ -111,6 +111,7 @@ def _get_coredump_path():
 
 class DebuggerThreadQemu(threading.Thread):
     """Debugger thread using QEMU and AFL fork server."""
+
     fork_server = None  # use class attribute due to the procmon's behavior of creating a new debugger thread on restart
 
     def __init__(