ClusterFuzz的bot源码(fuzz engine的选择与调度之libfuzzer)阅读

def run():
  """Initialise builtin fuzzing engines."""
  engine.register('afl', afl_engine.AFLEngine)
  engine.register('blackbox', blackbox_engine.BlackboxEngine)
  engine.register('honggfuzz', honggfuzz_engine.HonggfuzzEngine)
  engine.register('libFuzzer', libFuzzer_engine.LibFuzzerEngine)
  engine.register('syzkaller', syzkaller_engine.SyzkallerEngine)

获取到任务->fuzz_task.py中的execute_task->FuzzingSession->run()->engine.get获取具体的引擎类，调用do_engine_fuzzing(engine_impl)  ->  run_engine_fuzzer(engine_impl, self.fuzz_target.binary, sync_corpus_directory, self.testcase_directory) ->

options = engine_impl.prepare(sync_corpus_directory, target_path, build_dir)
fuzz_test_timeout = environment.get_value('FUZZ_TEST_TIMEOUT')
  additional_processing_time = engine_impl.fuzz_additional_processing_timeout(
      options)
......
......
......
result = engine_impl.fuzz(target_path, options, testcase_directory,
                            fuzz_test_timeout)

AFLEngine
BlackboxEngine
HonggfuzzEngine
LibFuzzerEngine
SyzkallerEngine

arguments = fuzzer.get_arguments(target_path)

grammar = fuzzer.get_grammar(target_path)

strategy_pool = strategy_selection.generate_weighted_strategy_pool(
        strategy_list=strategy.LIBFUZZER_STRATEGY_LIST,
        use_generator=True,
        engine_name=self.name)
strategy_info = libfuzzer.pick_strategies(strategy_pool, target_path,
                                              corpus_dir, arguments, grammar)

LIBFUZZER_STRATEGY_LIST = [
    CORPUS_MUTATION_RADAMSA_STRATEGY,
    RANDOM_MAX_LENGTH_STRATEGY,
    CORPUS_MUTATION_ML_RNN_STRATEGY,
    VALUE_PROFILE_STRATEGY,
    FORK_STRATEGY,
    CORPUS_SUBSET_STRATEGY,
    RECOMMENDED_DICTIONARY_STRATEGY,
    DATAFLOW_TRACING_STRATEGY,
    MUTATOR_PLUGIN_STRATEGY,
    MUTATOR_PLUGIN_RADAMSA_STRATEGY,
    PEACH_GRAMMAR_MUTATION_STRATEGY,
]

distribution = environment.get_value('STRATEGY_SELECTION_DISTRIBUTION')

strategy_selection = utils.random_weighted_choice(distribution_tuples,
                                                    'probability')

return generate_default_strategy_pool(strategy_list, use_generator)

def generate_default_strategy_pool(strategy_list, use_generator):
  """Return a strategy pool representing a selection of strategies for launcher
  to consider.

  Select strategies according to default strategy selection method."""
  pool = StrategyPool()

  # If use_generator is enabled, decide whether to include radamsa, ml rnn,
  # or no generator (mutually exclusive).
  if use_generator:
    choose_generator(pool)

  # Decide whether or not to add non-generator strategies according to
  # probability parameters.
  for value in [
      strategy_entry for strategy_entry in strategy_list
      if strategy_entry not in GENERATORS
  ]:
    if do_strategy(value):
      pool.add_strategy(value)

  logs.log('Strategy pool was generated according to default parameters. '
           'Chosen strategies: ' + ', '.join(pool.strategy_names))
  return pool

# /src/python/bot/fuzzers/engine_common.py
def decide_with_probability(probability):
  """Decide if we want to do something with the given probability."""
  return random.SystemRandom().random() < probability
  
# /src/python/bot/fuzzers/strategy_selection.py
 def choose_generator(strategy_pool):
  """Chooses whether to use radamsa, ml rnn, or no generator and updates the
  strategy pool."""

  radamsa_prob = engine_common.get_strategy_probability(
      strategy.CORPUS_MUTATION_RADAMSA_STRATEGY.name,
      default=strategy.CORPUS_MUTATION_RADAMSA_STRATEGY.probability)

  ml_rnn_prob = engine_common.get_strategy_probability(
      strategy.CORPUS_MUTATION_ML_RNN_STRATEGY.name,
      default=strategy.CORPUS_MUTATION_ML_RNN_STRATEGY.probability)

  if engine_common.decide_with_probability(radamsa_prob + ml_rnn_prob):
    if engine_common.decide_with_probability(
        radamsa_prob / (radamsa_prob + ml_rnn_prob)):
      strategy_pool.add_strategy(strategy.CORPUS_MUTATION_RADAMSA_STRATEGY)
    else:
      strategy_pool.add_strategy(strategy.CORPUS_MUTATION_ML_RNN_STRATEGY)

StrategyInfo(fuzzing_strategies, arguments, additional_corpus_dirs,
                      extra_env, use_dataflow_tracing, is_mutations_run)

# Depends on the presense of DFSan instrumented build.
  dataflow_build_dir = environment.get_value('DATAFLOW_BUILD_DIR')
  use_dataflow_tracing = (
      dataflow_build_dir and
      strategy_pool.do_strategy(strategy.DATAFLOW_TRACING_STRATEGY))
  if use_dataflow_tracing:
    dataflow_binary_path = os.path.join(
        dataflow_build_dir, os.path.relpath(fuzzer_path, build_directory))
    dataflow_trace_dir = dataflow_binary_path + DATAFLOW_TRACE_DIR_SUFFIX
    if os.path.exists(dataflow_trace_dir):
      arguments.append(
          '%s%s' % (constants.DATA_FLOW_TRACE_FLAG, dataflow_trace_dir))
      arguments.append('%s%s' % (constants.FOCUS_FUNCTION_FLAG, 'auto'))
      fuzzing_strategies.append(strategy.DATAFLOW_TRACING_STRATEGY.name)
    else:
      logs.log_warn(
          'Dataflow trace is not found in dataflow build, skipping strategy.')
      use_dataflow_tracing = False

# Select a generator to attempt to use for existing testcase mutations.
  candidate_generator = engine_common.select_generator(strategy_pool,
                                                       fuzzer_path)
  is_mutations_run = (not environment.is_ephemeral() and
                      candidate_generator != engine_common.Generator.NONE)
					  

	# Generate new testcase mutations using radamsa, etc.
  if is_mutations_run:
    new_testcase_mutations_directory = create_corpus_directory('mutations')
    generator_used = engine_common.generate_new_testcase_mutations(
        corpus_directory, new_testcase_mutations_directory,
        project_qualified_fuzzer_name, candidate_generator)

    # Add the used generator strategy to our fuzzing strategies list.
    if generator_used:
      if candidate_generator == engine_common.Generator.RADAMSA:
        fuzzing_strategies.append(
            strategy.CORPUS_MUTATION_RADAMSA_STRATEGY.name)
      elif candidate_generator == engine_common.Generator.ML_RNN:
        fuzzing_strategies.append(strategy.CORPUS_MUTATION_ML_RNN_STRATEGY.name)

    additional_corpus_dirs.append(new_testcase_mutations_directory)

def generate_new_testcase_mutations(corpus_directory,
                                    new_testcase_mutations_directory,
                                    fuzzer_name, candidate_generator):
  """Generate new testcase mutations, using existing corpus directory or other
  methods.

  Returns true if mutations are successfully generated using radamsa or ml rnn.
  A false return signifies either no generator use or unsuccessful generation of
  testcase mutations."""
  generation_timeout = get_new_testcase_mutations_timeout()
  pre_mutations_filecount = shell.get_directory_file_count(
      new_testcase_mutations_directory)

  # Generate new testcase mutations using Radamsa.
  if candidate_generator == Generator.RADAMSA:
    generate_new_testcase_mutations_using_radamsa(
        corpus_directory, new_testcase_mutations_directory, generation_timeout)
  # Generate new testcase mutations using ML RNN model.
  elif candidate_generator == Generator.ML_RNN:
    generate_new_testcase_mutations_using_ml_rnn(
        corpus_directory, new_testcase_mutations_directory, fuzzer_name,
        generation_timeout)

  # If new mutations are successfully generated, return true.
  if shell.get_directory_file_count(
      new_testcase_mutations_directory) > pre_mutations_filecount:
    return True

  return False

if strategy_pool.do_strategy(strategy.RANDOM_MAX_LENGTH_STRATEGY):
    max_len_argument = fuzzer_utils.extract_argument(
        existing_arguments, constants.MAX_LEN_FLAG, remove=False)
    if not max_len_argument:
      max_length = random.SystemRandom().randint(1, MAX_VALUE_FOR_MAX_LENGTH)
      arguments.append('%s%d' % (constants.MAX_LEN_FLAG, max_length))
      fuzzing_strategies.append(strategy.RANDOM_MAX_LENGTH_STRATEGY.name)

if (strategy_pool.do_strategy(strategy.RECOMMENDED_DICTIONARY_STRATEGY) and
      add_recommended_dictionary(arguments, project_qualified_fuzzer_name,
                                 fuzzer_path)):
    fuzzing_strategies.append(strategy.RECOMMENDED_DICTIONARY_STRATEGY.name)

def add_recommended_dictionary(arguments, fuzzer_name, fuzzer_path):
  """Add recommended dictionary from GCS to existing .dict file or create
  a new one and update the arguments as needed.
  This function modifies |arguments| list in some cases."""
  recommended_dictionary_path = os.path.join(
      fuzzer_utils.get_temp_dir(),
      dictionary_manager.RECOMMENDED_DICTIONARY_FILENAME)

  dict_manager = dictionary_manager.DictionaryManager(fuzzer_name)

  try:
    # Bail out if cannot download recommended dictionary from GCS.
    if not dict_manager.download_recommended_dictionary_from_gcs(
        recommended_dictionary_path):
      return False
  except Exception as ex:
    logs.log_error(
        'Exception downloading recommended dictionary:\n%s.' % str(ex))
    return False

  # Bail out if the downloaded dictionary is empty.
  if not os.path.getsize(recommended_dictionary_path):
    return False

  # Check if there is an existing dictionary file in arguments.
  original_dictionary_path = fuzzer_utils.extract_argument(
      arguments, constants.DICT_FLAG)
  merged_dictionary_path = (
      original_dictionary_path or
      dictionary_manager.get_default_dictionary_path(fuzzer_path))
  merged_dictionary_path += MERGED_DICT_SUFFIX

  dictionary_manager.merge_dictionary_files(original_dictionary_path,
                                            recommended_dictionary_path,
                                            merged_dictionary_path)
  arguments.append(constants.DICT_FLAG + merged_dictionary_path)
  return True

if strategy_pool.do_strategy(strategy.VALUE_PROFILE_STRATEGY):
    arguments.append(constants.VALUE_PROFILE_ARGUMENT)
    fuzzing_strategies.append(strategy.VALUE_PROFILE_STRATEGY.name)

# Do not use fork mode for DFT-based fuzzing. This is needed in order to
 # collect readable and actionable logs from fuzz targets running with DFT.
 if (not is_fuchsia and not is_android and not is_ephemeral and
     not use_dataflow_tracing and
     strategy_pool.do_strategy(strategy.FORK_STRATEGY)):
   max_fuzz_threads = environment.get_value('MAX_FUZZ_THREADS', 1)
   num_fuzz_processes = max(1, utils.cpu_count() // max_fuzz_threads)
   arguments.append('%s%d' % (constants.FORK_FLAG, num_fuzz_processes))
   fuzzing_strategies.append(
       '%s_%d' % (strategy.FORK_STRATEGY.name, num_fuzz_processes))

extra_env = {}
if (strategy_pool.do_strategy(strategy.MUTATOR_PLUGIN_STRATEGY) and
    use_mutator_plugin(target_name, extra_env)):
  fuzzing_strategies.append(strategy.MUTATOR_PLUGIN_STRATEGY.name)

if (not has_existing_mutator_strategy(fuzzing_strategies) and
      strategy_pool.do_strategy(strategy.PEACH_GRAMMAR_MUTATION_STRATEGY) and
      use_peach_mutator(extra_env, grammar)):
    fuzzing_strategies.append(
        '%s_%s' % (strategy.PEACH_GRAMMAR_MUTATION_STRATEGY.name, grammar))

MUTATOR_STRATEGIES = [
    strategy.PEACH_GRAMMAR_MUTATION_STRATEGY.name,
    strategy.MUTATOR_PLUGIN_STRATEGY.name,
    strategy.MUTATOR_PLUGIN_RADAMSA_STRATEGY.name
]

# Set title and pit environment variables
  extra_env['PIT_FILENAME'] = pit_path
  extra_env['PIT_TITLE'] = grammar
# Set LD_PRELOAD.
  peach_path = os.path.join(unzipped, 'peach_mutator', 'src', 'peach.so')
  extra_env['LD_PRELOAD'] = peach_path
# Set Python path.
  new_path = [
      os.path.join(unzipped, 'peach_mutator', 'src'),
      os.path.join(unzipped, 'peach_mutator', 'third_party', 'peach'),
  ] + sys.path

  extra_env['PYTHONPATH'] = os.pathsep.join(new_path)

if (not has_existing_mutator_strategy(fuzzing_strategies) and
      strategy_pool.do_strategy(strategy.MUTATOR_PLUGIN_RADAMSA_STRATEGY) and
      use_radamsa_mutator_plugin(extra_env)):
    fuzzing_strategies.append(strategy.MUTATOR_PLUGIN_RADAMSA_STRATEGY.name)

def use_radamsa_mutator_plugin(extra_env):
  """Decide whether to use Radamsa in process. If yes, add the path to the
  radamsa shared object to LD_PRELOAD in |extra_env| and return True."""
  # Radamsa will only work on LINUX ASAN jobs.
  # TODO(mpherman): Include architecture info in job definition and exclude
  # i386.
  if environment.is_lib() or not is_linux_asan():
    return False

  radamsa_path = os.path.join(environment.get_platform_resources_directory(),
                              'radamsa', 'libradamsa.so')

  logs.log('Using Radamsa mutator plugin : %s' % radamsa_path)
  extra_env['LD_PRELOAD'] = radamsa_path
  return True

arguments.extend(strategy_info.arguments)

# Check for seed corpus and add it into corpus directory.
engine_common.unpack_seed_corpus_if_needed(target_path, corpus_dir)

# Pick a few testcases from our corpus to use as the initial corpus.
subset_size = engine_common.random_choice(
    engine_common.CORPUS_SUBSET_NUM_TESTCASES)

if (not strategy_info.use_dataflow_tracing and
    strategy_pool.do_strategy(strategy.CORPUS_SUBSET_STRATEGY) and
    shell.get_directory_file_count(corpus_dir) > subset_size):
  # Copy |subset_size| testcases into 'subset' directory.
  corpus_subset_dir = self._create_temp_corpus_dir('subset')
  libfuzzer.copy_from_corpus(corpus_subset_dir, corpus_dir, subset_size)
  strategy_info.fuzzing_strategies.append(
      strategy.CORPUS_SUBSET_STRATEGY.name + '_' + str(subset_size))
  strategy_info.additional_corpus_dirs.append(corpus_subset_dir)
else:
  strategy_info.additional_corpus_dirs.append(corpus_dir)

# Check dict argument to make sure that it's valid.
   dict_path = fuzzer_utils.extract_argument(
       arguments, constants.DICT_FLAG, remove=False)
   if dict_path and not os.path.exists(dict_path):
     logs.log_error('Invalid dict %s for %s.' % (dict_path, target_path))
     fuzzer_utils.extract_argument(arguments, constants.DICT_FLAG)
  
# If there's no dict argument, check for %target_binary_name%.dict file.
   dict_path = fuzzer_utils.extract_argument(
       arguments, constants.DICT_FLAG, remove=False)
   if not dict_path:
     dict_path = dictionary_manager.get_default_dictionary_path(target_path)
     if os.path.exists(dict_path):
       arguments.append(constants.DICT_FLAG + dict_path)
	
# If we have a dictionary, correct any items that are not formatted properly
   # (e.g. quote items that are missing them).
   dictionary_manager.correct_if_needed(dict_path)

strategies = stats.process_strategies(
       strategy_info.fuzzing_strategies, name_modifier=lambda x: x)
   return LibFuzzerOptions(
       corpus_dir, arguments, strategies, strategy_info.additional_corpus_dirs,
       strategy_info.extra_env, strategy_info.use_dataflow_tracing,
       strategy_info.is_mutations_run)

fuzz_test_timeout = environment.get_value('FUZZ_TEST_TIMEOUT')
additional_processing_time = engine_impl.fuzz_additional_processing_timeout(
    options)
fuzz_test_timeout -= additional_processing_time
if fuzz_test_timeout <= 0:
  raise FuzzTaskException(
      f'Invalid engine timeout: '
      f'{fuzz_test_timeout} - {additional_processing_time}')

result = engine_impl.fuzz(target_path, options, testcase_directory,
                            fuzz_test_timeout)

profiler.start_if_needed('libfuzzer_fuzz')
   runner = libfuzzer.get_runner(target_path)
   libfuzzer.set_sanitizer_options(target_path, fuzz_options=options)

   # Directory to place new units.
   new_corpus_dir = self._create_temp_corpus_dir('new')

   corpus_directories = [new_corpus_dir] + options.fuzz_corpus_dirs
   fuzz_result = runner.fuzz(
       corpus_directories,
       fuzz_timeout=max_time,
       additional_args=options.arguments,
       artifact_prefix=reproducers_dir,
       extra_env=options.extra_env)

log_lines = fuzz_result.output.splitlines()
   # Output can be large, so save some memory by removing reference to the
   # original output which is no longer needed.
   fuzz_result.output = None

   # Check if we crashed, and get the crash testcase path.
   crash_testcase_file_path = runner.get_testcase_path(log_lines)

   # If we exited with a non-zero return code with no crash file in output from
   # libFuzzer, this is most likely a startup crash. Use an empty testcase to
   # to store it as a crash.
   if not crash_testcase_file_path and fuzz_result.return_code:
     crash_testcase_file_path = self._create_empty_testcase_file(
         reproducers_dir)

   # Parse stats information based on libFuzzer output.
   parsed_stats = libfuzzer.parse_log_stats(log_lines)

   # Extend parsed stats by additional performance features.
   parsed_stats.update(
       stats.parse_performance_features(log_lines, options.strategies,
                                        options.arguments))

   # Set some initial stat overrides.
   timeout_limit = fuzzer_utils.extract_argument(
       options.arguments, constants.TIMEOUT_FLAG, remove=False)

   expected_duration = runner.get_max_total_time(max_time)
   actual_duration = int(fuzz_result.time_executed)
   fuzzing_time_percent = 100 * actual_duration / float(expected_duration)
   parsed_stats.update({
       'timeout_limit': int(timeout_limit),
       'expected_duration': expected_duration,
       'actual_duration': actual_duration,
       'fuzzing_time_percent': fuzzing_time_percent,
   })

   # Remove fuzzing arguments before merge and dictionary analysis step.
   merge_arguments = options.arguments[:]
   libfuzzer.remove_fuzzing_arguments(merge_arguments, is_merge=True)
   self._merge_new_units(target_path, options.corpus_dir, new_corpus_dir,
                         options.fuzz_corpus_dirs, merge_arguments,
                         parsed_stats)

   fuzz_logs = '\n'.join(log_lines)
   crashes = []
   if crash_testcase_file_path:
     reproduce_arguments = options.arguments[:]
     libfuzzer.remove_fuzzing_arguments(reproduce_arguments)

     # Use higher timeout for reproduction.
     libfuzzer.fix_timeout_argument_for_reproduction(reproduce_arguments)

     # Write the new testcase.
     # Copy crash testcase contents into the main testcase path.
     crashes.append(
         engine.Crash(crash_testcase_file_path, fuzz_logs, reproduce_arguments,
                      actual_duration))

   libfuzzer.analyze_and_update_recommended_dictionary(
       runner, project_qualified_fuzzer_name, log_lines, options.corpus_dir,
       merge_arguments)

   return engine.FuzzResult(fuzz_logs, fuzz_result.command, crashes,
                            parsed_stats, fuzz_result.time_executed)

class LibFuzzerRunner(new_process.UnicodeProcessRunner, LibFuzzerCommon):
  """libFuzzer runner (when minijail is not used)."""

  def __init__(self, executable_path, default_args=None):
    """Inits the LibFuzzerRunner.

    Args:
      executable_path: Path to the fuzzer executable.
      default_args: Default arguments to always pass to the fuzzer.
    """
    super().__init__(executable_path=executable_path, default_args=default_args)

  def fuzz(self,
           corpus_directories,
           fuzz_timeout,
           artifact_prefix=None,
           additional_args=None,
           extra_env=None):
    """LibFuzzerCommon.fuzz override."""
    additional_args = copy.copy(additional_args)
    if additional_args is None:
      additional_args = []

    return LibFuzzerCommon.fuzz(self, corpus_directories, fuzz_timeout,
                                artifact_prefix, additional_args, extra_env)

def fuzz(self,
           corpus_directories,
           fuzz_timeout,
           artifact_prefix=None,
           additional_args=None,
           extra_env=None):
    """Running fuzzing command.

    Args:
      corpus_directories: List of corpus directory paths to be passed to
          libFuzzer.
      fuzz_timeout: The maximum time in seconds that libFuzzer is allowed to run
          for.
      artifact_prefix: The directory to store new fuzzing artifacts (crashes,
          timeouts, slow units)
      additional_args: A sequence of additional arguments to be passed to the
          executable.
      extra_env: A dictionary containing environment variables and their values.
          These will be added to the environment of the new process.

    Returns:
      A process.ProcessResult.
    """
    additional_args = copy.copy(additional_args)
    if additional_args is None:
      additional_args = []

    max_total_time = self.get_max_total_time(fuzz_timeout)
    if any(arg.startswith(constants.FORK_FLAG) for arg in additional_args):
      max_total_time -= self.LIBFUZZER_FORK_MODE_CLEAN_EXIT_TIME
    assert max_total_time > 0

    # Old libFuzzer jobs specify -artifact_prefix through additional_args
    if artifact_prefix:
      additional_args.append(
          '%s%s' % (constants.ARTIFACT_PREFIX_FLAG,
                    self._normalize_artifact_prefix(artifact_prefix)))

    additional_args.extend([
        '%s%d' % (constants.MAX_TOTAL_TIME_FLAG, max_total_time),
        constants.PRINT_FINAL_STATS_ARGUMENT,
        # FIXME: temporarily disabled due to a lack of crash information in
        # output.
        # '-close_fd_mask=3',
    ])

    additional_args.extend(corpus_directories)
    return self.run_and_wait(
        additional_args=additional_args,
        timeout=fuzz_timeout - self.SIGTERM_WAIT_TIME,
        terminate_before_kill=True,
        terminate_wait_time=self.SIGTERM_WAIT_TIME,
        max_stdout_len=MAX_OUTPUT_LEN,
        extra_env=extra_env)