[Centipede] Perform corpus minimization during a fuzzing session (#4706)

This change call's centipede's minimize_corpus method on each fuzzing
round.
This allows us to distill the corpus and only add useful units.
On a local run, this allowed to decrease the number of units added from
2000+ to 5 with the same coverage.
The cl also extracts some common functionality into engine_common.

Author: alhijazi

src/clusterfuzz/_internal/bot/fuzzers/centipede/constants.py

@@ -39,6 +39,10 @@
 EXTRA_BINARIES_FLAGNAME = 'extra_binaries'
 EXIT_ON_CRASH_FLAGNAME = 'exit_on_crash'
 
+MAX_LEN_FLAGNAME = 'max_len'
+NUM_RUNS_FLAGNAME = 'num_runs'
+BATCH_SIZE_FLAGNAME = 'batch_size'
+
 NUM_RUNS_PER_MINIMIZATION = 100000
 
 

src/clusterfuzz/_internal/bot/fuzzers/centipede/engine.py

@@ -32,6 +32,7 @@
 from clusterfuzz._internal.metrics import logs
 from clusterfuzz._internal.system import environment
 from clusterfuzz._internal.system import new_process
+from clusterfuzz._internal.system import shell
 from clusterfuzz.fuzz import engine
 from clusterfuzz.stacktraces import constants as stacktraces_constants
 
@@ -46,6 +47,17 @@ class CentipedeError(Exception):
   """Base exception class."""
 
 
+class CentipedeOptions(engine.FuzzOptions):
+  """Centipede engine options."""
+
+  def __init__(self, corpus_dir, arguments, strategies, workdir,
+               new_corpus_dir):
+    super().__init__(corpus_dir, arguments, strategies)
+    # Directory to add new units
+    self.new_corpus_dir = new_corpus_dir
+    self.workdir = workdir
+
+
 def _get_runner(target_path):
   """Gets the Centipede runner."""
   centipede_path = pathlib.Path(target_path).parent / 'centipede'
@@ -198,11 +210,13 @@ def prepare(self, corpus_dir, target_path, build_dir):
     # 1. Centipede-readable corpus file;
     # 2. Centipede-readable feature file;
     # 3. Crash reproducing inputs.
-    workdir = self._create_temp_dir('workdir')
+    workdir = engine_common.create_temp_fuzzing_dir('workdir')
     arguments[constants.WORKDIR_FLAGNAME] = str(workdir)
 
-    # Directory corpus_dir saves the corpus files required by ClusterFuzz.
-    arguments[constants.CORPUS_DIR_FLAGNAME] = corpus_dir
+    # Directory to place new units. While fuzzing, the new corpus
+    # elements are written to the first dir in the list of corpus directories.
+    new_corpus_dir = engine_common.create_temp_fuzzing_dir('new')
+    arguments[constants.CORPUS_DIR_FLAGNAME] = f'{new_corpus_dir},{corpus_dir}'
 
     target_binaries = self._get_binary_paths(target_path)
     if target_binaries.unsanitized is None:
@@ -214,7 +228,8 @@ def prepare(self, corpus_dir, target_path, build_dir):
       arguments[constants.EXTRA_BINARIES_FLAGNAME] = str(
           target_binaries.sanitized)
 
-    return engine.FuzzOptions(corpus_dir, arguments.list(), {})
+    return CentipedeOptions(corpus_dir, arguments.list(), {}, workdir,
+                            new_corpus_dir)
 
   def _get_binary_paths(self, target_path):
     """Gets the paths to the main and auxiliary binaries based on |target_path|
@@ -284,11 +299,42 @@ def fuzz(self, target_path, options, reproducers_dir, max_time):  # pylint: disa
     runner = _get_runner(target_path)
     _set_sanitizer_options(target_path)
     timeout = max_time + _CLEAN_EXIT_SECS
+
+    old_corpus_len = shell.get_directory_file_count(options.corpus_dir)
+    logs.info(f'Corpus length before fuzzing: {old_corpus_len}')
+
     fuzz_result = runner.run_and_wait(
         additional_args=options.arguments, timeout=timeout)
     log_lines = fuzz_result.output.splitlines()
     fuzz_result.output = Engine.trim_logs(fuzz_result.output)
 
+    workdir = options.workdir
+
+    try:
+      time_for_minimize = timeout - fuzz_result.time_executed
+
+      self.minimize_corpus(
+          target_path=target_path,
+          arguments=[],
+          # New units, in addition to the main corpus units,
+          # are placed in new_corpus_dir. Minimize and merge back
+          # to the main corpus_dir.
+          input_dirs=[options.new_corpus_dir],
+          output_dir=options.corpus_dir,
+          reproducers_dir=reproducers_dir,
+          max_time=time_for_minimize,
+          # Use the same workdir that was used for fuzzing.
+          # This allows us to skip rerunning the fuzzing inputs.
+          workdir=workdir)
+    except:
+      # TODO(alhijazi): Convert to a warning if this becomes a problem
+      # caused by user code rather than by ClusterFuzz or Centipede.
+      logs.error('Corpus minimization failed.')
+      # If we fail to minimize, fall back to moving the new units
+      # from the new corpus_dir to the main corpus_dir.
+      engine_common.move_mergeable_units(options.new_corpus_dir,
+                                         options.corpus_dir)
+
     reproducer_path = _get_reproducer_path(fuzz_result.output, reproducers_dir)
     crashes = []
     if reproducer_path:
@@ -298,11 +344,7 @@ def fuzz(self, target_path, options, reproducers_dir, max_time):  # pylint: disa
               int(fuzz_result.time_executed)))
 
     stats_filename = f'fuzzing-stats-{os.path.basename(target_path)}.000000.csv'
-    args = fuzzer_options.FuzzerArguments.from_list(options.arguments)
-    assert args is not None
-    assert constants.WORKDIR_FLAGNAME in args
 
-    workdir = args[constants.WORKDIR_FLAGNAME]
     stats_file = os.path.join(workdir, stats_filename)
     stats = _parse_centipede_stats(stats_file)
     if not stats:
@@ -321,6 +363,11 @@ def fuzz(self, target_path, options, reproducers_dir, max_time):  # pylint: disa
     num_execs_avg = stats.get('NumExecs_Avg', 0.0)
     stats['average_exec_per_sec'] = num_execs_avg / fuzz_time_secs_avg
     stats.update(_parse_centipede_logs(log_lines))
+
+    new_corpus_len = shell.get_directory_file_count(options.corpus_dir)
+    logs.info(f'Corpus length after fuzzing: {new_corpus_len}')
+    new_units_added = new_corpus_len - old_corpus_len
+    stats['new_units_added'] = new_units_added
     return engine.FuzzResult(fuzz_result.output, fuzz_result.command, crashes,
                              stats, fuzz_result.time_executed)
 
@@ -379,14 +426,28 @@ def reproduce(self, target_path, input_path, arguments, max_time):  # pylint: di
     return engine.ReproduceResult(result.command, result.return_code,
                                   result.time_executed, result.output)
 
-  def _create_temp_dir(self, name):
-    """Creates temporary directory for fuzzing."""
-    new_directory = pathlib.Path(fuzzer_utils.get_temp_dir(), name)
-    engine_common.recreate_directory(new_directory)
-    return new_directory
+  def _strip_fuzzing_arguments(self, arguments):
+    """Remove arguments only needed for fuzzing."""
+    for argument in [
+        constants.FORK_SERVER_FLAGNAME,
+        constants.MAX_LEN_FLAGNAME,
+        constants.NUM_RUNS_FLAGNAME,
+        constants.EXIT_ON_CRASH_FLAGNAME,
+        constants.BATCH_SIZE_FLAGNAME,
+    ]:
+      if argument in arguments:
+        del arguments[argument]
+
+    return arguments
 
-  def minimize_corpus(self, target_path, arguments, input_dirs, output_dir,
-                      reproducers_dir, max_time):
+  def minimize_corpus(self,
+                      target_path,
+                      arguments,
+                      input_dirs,
+                      output_dir,
+                      reproducers_dir,
+                      max_time,
+                      workdir=None):
     """Runs corpus minimization.
     Args:
       target_path: Path to the target.
@@ -401,16 +462,29 @@ def minimize_corpus(self, target_path, arguments, input_dirs, output_dir,
       A FuzzResult object.
     """
     runner = _get_runner(target_path)
+    _set_sanitizer_options(target_path)
+
+    minimize_arguments = self._get_arguments(target_path)
+    self._strip_fuzzing_arguments(minimize_arguments)
+    environment.set_value('ASAN_OPTIONS', 'detect_odr_violation=0')
 
     # Step 1: Generate corpus file for Centipede.
-    full_corpus_workdir = self._create_temp_dir('full_corpus_workdir')
+    # When calling this during a fuzzing session, use the existing workdir.
+    # This avoids us having to re-run inputs and waste time unnecessarily.
+    # This saves a lot of time when the input corpus contains thousands
+    # of files.
+    full_corpus_workdir = workdir
+    if not full_corpus_workdir:
+      full_corpus_workdir = engine_common.create_temp_fuzzing_dir(
+          'full_corpus_workdir')
     input_dirs_param = ','.join(str(dir) for dir in input_dirs)
-    args = [
+    args = minimize_arguments.list() + [
         f'--workdir={full_corpus_workdir}',
         f'--binary={target_path}',
         f'--corpus_dir={input_dirs_param}',
         '--num_runs=0',
     ]
+    logs.info(f'Running Generate Corpus file for Centipede with args: {args}')
     result = runner.run_and_wait(additional_args=args, timeout=max_time)
     max_time -= result.time_executed
 
@@ -422,11 +496,12 @@ def minimize_corpus(self, target_path, arguments, input_dirs, output_dir,
       raise TimeoutError('Minimization timed out.')
 
     # Step 2: Distill.
-    args = [
+    args = minimize_arguments.list() + [
         f'--workdir={full_corpus_workdir}',
         f'--binary={target_path}',
-        '--distill',
+        '--distill=true',
     ]
+    logs.info(f'Running Corpus Distillation with args: {args}')
     result = runner.run_and_wait(additional_args=args, timeout=max_time)
     max_time -= result.time_executed
 
@@ -438,17 +513,21 @@ def minimize_corpus(self, target_path, arguments, input_dirs, output_dir,
 
     # Step 3: Generate corpus files for output_dir.
     os.makedirs(output_dir, exist_ok=True)
-    minimized_corpus_workdir = self._create_temp_dir('minimized_corpus_workdir')
+    minimized_corpus_workdir = engine_common.create_temp_fuzzing_dir(
+        'minimized_corpus_workdir')
+    logs.info(f'Created a temporary minimized corpus '
+              f'workdir {minimized_corpus_workdir}')
     distilled_file = os.path.join(
         full_corpus_workdir,
         f'distilled-{os.path.basename(target_path)}.000000')
     corpus_file = os.path.join(minimized_corpus_workdir, 'corpus.000000')
     shutil.copyfile(distilled_file, corpus_file)
 
-    args = [
+    args = minimize_arguments.list() + [
         f'--workdir={minimized_corpus_workdir}',
         f'--corpus_to_files={output_dir}',
     ]
+    logs.info(f'Converting corpus to files with the following args: {args}')
     result = runner.run_and_wait(additional_args=args, timeout=max_time)
 
     if result.timed_out or max_time < 0:
@@ -461,11 +540,16 @@ def minimize_corpus(self, target_path, arguments, input_dirs, output_dir,
     # Step 4: Copy reproducers from full_corpus_workdir.
     os.makedirs(reproducers_dir, exist_ok=True)
     crashes_dir = os.path.join(full_corpus_workdir, 'crashes')
-    for file in os.listdir(crashes_dir):
-      crasher_path = os.path.join(crashes_dir, file)
-      shutil.copy(crasher_path, reproducers_dir)
-    shutil.rmtree(full_corpus_workdir)
+
+    if os.path.exists(crashes_dir):
+      for file in os.listdir(crashes_dir):
+        crasher_path = os.path.join(crashes_dir, file)
+        shutil.copy(crasher_path, reproducers_dir)
+
     shutil.rmtree(minimized_corpus_workdir)
+    if not workdir:
+      # Only remove this directory if it was created in this method.
+      shutil.rmtree(full_corpus_workdir)
 
     return engine.ReproduceResult(result.command, result.return_code,
                                   result.time_executed, result.output)
@@ -507,7 +591,7 @@ def minimize_testcase(self, target_path, arguments, input_path, output_path,
       TimeoutError: If the testcase minimization exceeds max_time.
     """
     runner = _get_runner(target_path)
-    workdir = self._create_temp_dir('workdir')
+    workdir = engine_common.create_temp_fuzzing_dir('workdir')
     args = [
         f'--binary={target_path}',
         f'--workdir={workdir}',

src/clusterfuzz/_internal/bot/fuzzers/engine_common.py

@@ -21,6 +21,7 @@
 import re
 import shlex
 import shutil
+import string
 import sys
 import time
 
@@ -71,6 +72,8 @@
 # Maximum number of seconds to run radamsa for.
 RADAMSA_TIMEOUT = 3
 
+HEXDIGITS_SET = set(string.hexdigits)
+
 
 class Generator:
   """Generators we can use."""
@@ -656,3 +659,33 @@ def get_log_header(command, time_executed):
   """Get the log header."""
   quoted_command = get_command_quoted(command)
   return f'Command: {quoted_command}\nTime ran: {time_executed}\n'
+
+
+def is_sha1_hash(possible_hash):
+  """Returns True if |possible_hash| looks like a valid sha1 hash."""
+  if len(possible_hash) != 40:
+    return False
+
+  return all(char in HEXDIGITS_SET for char in possible_hash)
+
+
+def move_mergeable_units(merge_directory, corpus_directory):
+  """Move new units in |merge_directory| into |corpus_directory|."""
+  initial_units = {
+      os.path.basename(filename)
+      for filename in shell.get_files_list(corpus_directory)
+  }
+
+  for unit_path in shell.get_files_list(merge_directory):
+    unit_name = os.path.basename(unit_path)
+    if unit_name in initial_units and is_sha1_hash(unit_name):
+      continue
+    dest_path = os.path.join(corpus_directory, unit_name)
+    shell.move(unit_path, dest_path)
+
+
+def create_temp_fuzzing_dir(name):
+  """Create a temporary directory for fuzzing."""
+  new_corpus_directory = os.path.join(fuzzer_utils.get_temp_dir(), name)
+  recreate_directory(new_corpus_directory)
+  return new_corpus_directory

src/clusterfuzz/_internal/bot/fuzzers/honggfuzz/engine.py

@@ -21,7 +21,6 @@
 from clusterfuzz._internal.base import utils
 from clusterfuzz._internal.bot.fuzzers import dictionary_manager
 from clusterfuzz._internal.bot.fuzzers import engine_common
-from clusterfuzz._internal.bot.fuzzers import utils as fuzzer_utils
 from clusterfuzz._internal.metrics import logs
 from clusterfuzz._internal.system import environment
 from clusterfuzz._internal.system import new_process
@@ -219,12 +218,6 @@ def reproduce(self, target_path, input_path, arguments, max_time):  # pylint: di
     return engine.ReproduceResult(result.command, result.return_code,
                                   result.time_executed, result.output)
 
-  def _create_temp_corpus_dir(self, name):
-    """Creates temporary corpus directory."""
-    new_corpus_directory = os.path.join(fuzzer_utils.get_temp_dir(), name)
-    engine_common.recreate_directory(new_corpus_directory)
-    return new_corpus_directory
-
   def minimize_corpus(self, target_path, arguments, input_dirs, output_dir,
                       reproducers_dir, max_time):
     """Optional (but recommended): run corpus minimization.
@@ -244,7 +237,8 @@ def minimize_corpus(self, target_path, arguments, input_dirs, output_dir,
     del reproducers_dir
 
     runner = _get_runner()
-    combined_corpus_dir = self._create_temp_corpus_dir('minimize-workdir')
+    combined_corpus_dir = engine_common.create_temp_fuzzing_dir(
+        'minimize-workdir')
 
     # Copy all of the seeds into corpus.
     idx = 0