text: # Copyright 2021 Google LLC
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #      http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  """Module for running fuzzers."""
  import enum
  import logging
  import os
  import shutil
  import sys
  import time
  
  import clusterfuzz_deployment
  import config_utils
  import fuzz_target
  import generate_coverage_report
  import stack_parser
  
  # pylint: disable=wrong-import-position,import-error
  sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
  
  import utils
  
  
  class RunFuzzersResult(enum.Enum):
    """Enum result from running fuzzers."""
    ERROR = 0
    BUG_FOUND = 1
    NO_BUG_FOUND = 2
  
  
  class BaseFuzzTargetRunner:
    """Base class for fuzzer runners."""
  
    def __init__(self, config):
      self.config = config
      self.workspace = config_utils.Workspace(config)
      self.clusterfuzz_deployment = (
          clusterfuzz_deployment.get_clusterfuzz_deployment(
              self.config, self.workspace))
  
      # Set by the initialize method.
      self.fuzz_target_paths = None
  
    def get_fuzz_targets(self):
      """Returns fuzz targets in out directory."""
      return utils.get_fuzz_targets(self.workspace.out)
  
    def initialize(self):
      """Initialization method. Must be called before calling run_fuzz_targets.
      Returns True on success."""
      # Use a seperate initialization function so we can return False on failure
      # instead of exceptioning like we need to do if this were done in the
      # __init__ method.
  
      logging.info('Using %s sanitizer.', self.config.sanitizer)
  
      # TODO(metzman) Add a check to ensure we aren't over time limit.
      if not self.config.fuzz_seconds or self.config.fuzz_seconds < 1:
        logging.error(
            'Fuzz_seconds argument must be greater than 1, but was: %s.',
            self.config.fuzz_seconds)
        return False
  
      if not os.path.exists(self.workspace.out):
        logging.error('Out directory: %s does not exist.', self.workspace.out)
        return False
  
      if not os.path.exists(self.workspace.artifacts):
        os.makedirs(self.workspace.artifacts)
      elif (not os.path.isdir(self.workspace.artifacts) or
            os.listdir(self.workspace.artifacts)):
        logging.error('Artifacts path: %s exists and is not an empty directory.',
                      self.workspace.artifacts)
        return False
  
      self.fuzz_target_paths = self.get_fuzz_targets()
      logging.info('Fuzz targets: %s', self.fuzz_target_paths)
      if not self.fuzz_target_paths:
        logging.error('No fuzz targets were found in out directory: %s.',
                      self.workspace.out)
        return False
  
      return True
  
    def cleanup_after_fuzz_target_run(self, fuzz_target_obj):  # pylint: disable=no-self-use
      """Cleans up after running |fuzz_target_obj|."""
      raise NotImplementedError('Child class must implement method.')
  
    def run_fuzz_target(self, fuzz_target_obj):  # pylint: disable=no-self-use
      """Fuzzes with |fuzz_target_obj| and returns the result."""
      raise NotImplementedError('Child class must implement method.')
  
    @property
    def quit_on_bug_found(self):
      """Property that is checked to determine if fuzzing should quit after first
      bug is found."""
      raise NotImplementedError('Child class must implement method.')
  
    def get_fuzz_target_artifact(self, target, artifact_name):
      """Returns the path of a fuzzing artifact named |artifact_name| for
      |fuzz_target|."""
      artifact_name = (f'{target.target_name}-{self.config.sanitizer}-'
                       f'{artifact_name}')
      return os.path.join(self.workspace.artifacts, artifact_name)
  
    def create_fuzz_target_obj(self, target_path, run_seconds):
      """Returns a fuzz target object."""
      return fuzz_target.FuzzTarget(target_path, run_seconds, self.workspace,
                                    self.clusterfuzz_deployment, self.config)
  
    def run_fuzz_targets(self):
      """Runs fuzz targets. Returns True if a bug was found."""
      fuzzers_left_to_run = len(self.fuzz_target_paths)
  
      # Make a copy since we will mutate it.
      fuzz_seconds = self.config.fuzz_seconds
  
      min_seconds_per_fuzzer = fuzz_seconds // fuzzers_left_to_run
      bug_found = False
      for target_path in self.fuzz_target_paths:
        # By doing this, we can ensure that every fuzz target runs for at least
        # min_seconds_per_fuzzer, but that other fuzzers will have longer to run
        # if one ends early.
        run_seconds = max(fuzz_seconds // fuzzers_left_to_run,
                          min_seconds_per_fuzzer)
  
        target = self.create_fuzz_target_obj(target_path, run_seconds)
        start_time = time.time()
        result = self.run_fuzz_target(target)
        self.cleanup_after_fuzz_target_run(target)
  
        # It's OK if this goes negative since we take max when determining
        # run_seconds.
        fuzz_seconds -= time.time() - start_time
  
        fuzzers_left_to_run -= 1
        if not result.testcase or not result.stacktrace:
          logging.info('Fuzzer %s finished running without crashes.',
                       target.target_name)
          continue
  
        # TODO(metzman): Do this with filestore.
        testcase_artifact_path = self.get_fuzz_target_artifact(
            target, os.path.basename(result.testcase))
        shutil.move(result.testcase, testcase_artifact_path)
        bug_summary_artifact_path = self.get_fuzz_target_artifact(
            target, 'bug-summary.txt')
        stack_parser.parse_fuzzer_output(result.stacktrace,
                                         bug_summary_artifact_path)
  
        bug_found = True
        if self.quit_on_bug_found:
          logging.info('Bug found. Stopping fuzzing.')
          return bug_found
  
      return bug_found
  
  
  class CoverageTargetRunner(BaseFuzzTargetRunner):
    """Runner that runs the 'coverage' command."""
  
    @property
    def quit_on_bug_found(self):
      raise NotImplementedError('Not implemented for CoverageTargetRunner.')
  
    def get_fuzz_targets(self):
      """Returns fuzz targets in out directory."""
      # We only want fuzz targets from the root because during the coverage build,
      # a lot of the image's filesystem is copied into /out for the purpose of
      # generating coverage reports.
      # TOOD(metzman): Figure out if top_level_only should be the only behavior
      # for this function.
      return utils.get_fuzz_targets(self.workspace.out, top_level_only=True)
  
    def run_fuzz_targets(self):
      """Generates a coverage report. Always returns False since it never finds
      any bugs."""
      generate_coverage_report.generate_coverage_report(
          self.fuzz_target_paths, self.workspace, self.clusterfuzz_deployment,
          self.config)
      return False
  
    def run_fuzz_target(self, fuzz_target_obj):  # pylint: disable=no-self-use
      """Fuzzes with |fuzz_target_obj| and returns the result."""
      raise NotImplementedError('Not implemented for CoverageTargetRunner.')
  
    def cleanup_after_fuzz_target_run(self, fuzz_target_obj):  # pylint: disable=no-self-use
      """Cleans up after running |fuzz_target_obj|."""
      raise NotImplementedError('Not implemented for CoverageTargetRunner.')
  
  
  class CiFuzzTargetRunner(BaseFuzzTargetRunner):
    """Runner for fuzz targets used in CI (patch-fuzzing) context."""
  
    @property
    def quit_on_bug_found(self):
      return True
  
    def cleanup_after_fuzz_target_run(self, fuzz_target_obj):  # pylint: disable=no-self-use
      """Cleans up after running |fuzz_target_obj|."""
      fuzz_target_obj.free_disk_if_needed()
  
    def run_fuzz_target(self, fuzz_target_obj):  # pylint: disable=no-self-use
      return fuzz_target_obj.fuzz()
  
  
  class BatchFuzzTargetRunner(BaseFuzzTargetRunner):
    """Runner for fuzz targets used in batch fuzzing context."""
  
    @property
    def quit_on_bug_found(self):
      return False
  
    def run_fuzz_target(self, fuzz_target_obj):
      """Fuzzes with |fuzz_target_obj| and returns the result."""
      result = fuzz_target_obj.fuzz()
      logging.debug('corpus_path: %s', os.listdir(result.corpus_path))
      self.clusterfuzz_deployment.upload_corpus(fuzz_target_obj.target_name)
      return result
  
    def cleanup_after_fuzz_target_run(self, fuzz_target_obj):
      """Cleans up after running |fuzz_target_obj|."""
      # This must be done after we upload the corpus, otherwise it will be deleted
      # before we get a chance to upload it. We can't delete the fuzz target
      # because it is needed when we upload the build.
      fuzz_target_obj.free_disk_if_needed(delete_fuzz_target=False)
  
    def run_fuzz_targets(self):
      result = super().run_fuzz_targets()
  
      self.clusterfuzz_deployment.upload_crashes()
  
      # We want to upload the build to the filestore after we do batch fuzzing.
      # There are some is a problem with this. We don't want to upload the build
      # before fuzzing, because if we download the latest build, we will consider
      # the build we just uploaded to be the latest even though it shouldn't be
      # (we really intend to download the build before the curent one.
      # TODO(metzman): We should really be uploading latest build in build_fuzzers
      # before we remove unaffected fuzzers. Otherwise, we can lose fuzzers. This
      # is probably more of a theoretical concern since in batch fuzzing, there is
      # no code change and thus no fuzzers that are removed, but it's inelegant to
      # put this here.
  
      self.clusterfuzz_deployment.upload_latest_build()
      return result
  
  
  def get_fuzz_target_runner(config):
    """Returns a fuzz target runner object based on the run_fuzzers_mode of
    |config|."""
    logging.info('RUN_FUZZERS_MODE is: %s', config.run_fuzzers_mode)
    if config.run_fuzzers_mode == 'batch':
      return BatchFuzzTargetRunner(config)
    if config.run_fuzzers_mode == 'coverage':
      return CoverageTargetRunner(config)
    return CiFuzzTargetRunner(config)
  
  
  def run_fuzzers(config):  # pylint: disable=too-many-locals
    """Runs fuzzers for a specific OSS-Fuzz project.
  
    Args:
      config: A RunFuzzTargetsConfig.
  
    Returns:
      A RunFuzzersResult enum value indicating what happened during fuzzing.
    """
    fuzz_target_runner = get_fuzz_target_runner(config)
    if not fuzz_target_runner.initialize():
      # We didn't fuzz at all because of internal (CIFuzz) errors. And we didn't
      # find any bugs.
      return RunFuzzersResult.ERROR
  
    if not fuzz_target_runner.run_fuzz_targets():
      # We fuzzed successfully, but didn't find any bugs (in the fuzz target).
      return RunFuzzersResult.NO_BUG_FOUND
  
    # We fuzzed successfully and found bug(s) in the fuzz targets.
    return RunFuzzersResult.BUG_FOUND
  