# 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.
"""Tests for running fuzzers."""
import os
import sys
import shutil
import tempfile
import unittest
from unittest import mock

import parameterized
from pyfakefs import fake_filesystem_unittest

import config_utils
import fuzz_target
import run_fuzzers

# pylint: disable=wrong-import-position
INFRA_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(INFRA_DIR)

import test_helpers

# NOTE: This integration test relies on
# https://github.com/google/oss-fuzz/tree/master/projects/example project.
EXAMPLE_PROJECT = 'example'

# Location of files used for testing.
TEST_DATA_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                              'test_data')

MEMORY_FUZZER_DIR = os.path.join(TEST_DATA_PATH, 'memory')
MEMORY_FUZZER = 'curl_fuzzer_memory'

UNDEFINED_FUZZER_DIR = os.path.join(TEST_DATA_PATH, 'undefined')
UNDEFINED_FUZZER = 'curl_fuzzer_undefined'

FUZZ_SECONDS = 10


def _create_config(**kwargs):
  """Creates a config object and then sets every attribute that is a key in
  |kwargs| to the corresponding value. Asserts that each key in |kwargs| is an
  attribute of Config."""
  with mock.patch('os.path.basename', return_value=None), mock.patch(
      'config_utils.get_project_src_path',
      return_value=None), mock.patch('config_utils._is_dry_run',
                                     return_value=True):
    config = config_utils.RunFuzzersConfig()

  for key, value in kwargs.items():
    assert hasattr(config, key), 'Config doesn\'t have attribute: ' + key
    setattr(config, key, value)
  return config


class RunFuzzerIntegrationTestMixin:  # pylint: disable=too-few-public-methods,invalid-name
  """Mixin for integration test classes that runbuild_fuzzers on builds of a
  specific sanitizer."""
  # These must be defined by children.
  FUZZER_DIR = None
  FUZZER = None

  def _test_run_with_sanitizer(self, fuzzer_dir, sanitizer):
    """Calls run_fuzzers on fuzzer_dir and |sanitizer| and asserts
    the run succeeded and that no bug was found."""
    with test_helpers.temp_dir_copy(fuzzer_dir) as fuzzer_dir_copy:
      config = _create_config(fuzz_seconds=FUZZ_SECONDS,
                              workspace=fuzzer_dir_copy,
                              project_name='curl',
                              sanitizer=sanitizer)
      result = run_fuzzers.run_fuzzers(config)
    self.assertEqual(result, run_fuzzers.RunFuzzersResult.NO_BUG_FOUND)


class RunMemoryFuzzerIntegrationTest(RunFuzzerIntegrationTestMixin,
                                     unittest.TestCase):
  """Integration test for build_fuzzers with an MSAN build."""
  FUZZER_DIR = MEMORY_FUZZER_DIR
  FUZZER = MEMORY_FUZZER

  @unittest.skipIf(not os.getenv('INTEGRATION_TESTS'),
                   'INTEGRATION_TESTS=1 not set')
  def test_run_with_memory_sanitizer(self):
    """Tests run_fuzzers with a valid MSAN build."""
    self._test_run_with_sanitizer(self.FUZZER_DIR, 'memory')


class RunUndefinedFuzzerIntegrationTest(RunFuzzerIntegrationTestMixin,
                                        unittest.TestCase):
  """Integration test for build_fuzzers with an UBSAN build."""
  FUZZER_DIR = UNDEFINED_FUZZER_DIR
  FUZZER = UNDEFINED_FUZZER

  @unittest.skipIf(not os.getenv('INTEGRATION_TESTS'),
                   'INTEGRATION_TESTS=1 not set')
  def test_run_with_undefined_sanitizer(self):
    """Tests run_fuzzers with a valid UBSAN build."""
    self._test_run_with_sanitizer(self.FUZZER_DIR, 'undefined')


class BaseFuzzTargetRunnerTest(unittest.TestCase):
  """Tests BaseFuzzTargetRunner."""

  def _create_runner(self, **kwargs):  # pylint: disable=no-self-use
    defaults = {'fuzz_seconds': FUZZ_SECONDS, 'project_name': EXAMPLE_PROJECT}
    for default_key, default_value in defaults.items():
      if default_key not in kwargs:
        kwargs[default_key] = default_value

    config = _create_config(**kwargs)
    return run_fuzzers.BaseFuzzTargetRunner(config)

  def _test_initialize_fail(self, expected_error_args, **create_runner_kwargs):
    with mock.patch('logging.error') as mocked_error:
      runner = self._create_runner(**create_runner_kwargs)
      self.assertFalse(runner.initialize())
      mocked_error.assert_called_with(*expected_error_args)

  @parameterized.parameterized.expand([(0,), (None,), (-1,)])
  def test_initialize_invalid_fuzz_seconds(self, fuzz_seconds):
    """Tests initialize fails with an invalid fuzz seconds."""
    expected_error_args = ('Fuzz_seconds argument must be greater than 1, '
                           'but was: %s.', fuzz_seconds)
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'out')
      os.mkdir(out_path)
      with mock.patch('utils.get_fuzz_targets') as mocked_get_fuzz_targets:
        mocked_get_fuzz_targets.return_value = [
            os.path.join(out_path, 'fuzz_target')
        ]
        self._test_initialize_fail(expected_error_args,
                                   fuzz_seconds=fuzz_seconds,
                                   workspace=tmp_dir)

  def test_initialize_no_out_dir(self):
    """Tests initialize fails with no out dir."""
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'out')
      expected_error_args = ('Out directory: %s does not exist.', out_path)
      self._test_initialize_fail(expected_error_args, workspace=tmp_dir)

  def test_initialize_nonempty_artifacts(self):
    """Tests initialize with a file artifacts path."""
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'out')
      os.mkdir(out_path)
      artifacts_path = os.path.join(out_path, 'artifacts')
      with open(artifacts_path, 'w') as artifacts_handle:
        artifacts_handle.write('fake')
      expected_error_args = (
          'Artifacts path: %s exists and is not an empty directory.',
          artifacts_path)
      self._test_initialize_fail(expected_error_args, workspace=tmp_dir)

  def test_initialize_bad_artifacts(self):
    """Tests initialize with a non-empty artifacts path."""
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'out')
      artifacts_path = os.path.join(out_path, 'artifacts')
      os.makedirs(artifacts_path)
      artifact_path = os.path.join(artifacts_path, 'artifact')
      with open(artifact_path, 'w') as artifact_handle:
        artifact_handle.write('fake')
      expected_error_args = (
          'Artifacts path: %s exists and is not an empty directory.',
          artifacts_path)
      self._test_initialize_fail(expected_error_args, workspace=tmp_dir)

  @mock.patch('utils.get_fuzz_targets')
  @mock.patch('logging.error')
  def test_initialize_empty_artifacts(self, mocked_log_error,
                                      mocked_get_fuzz_targets):
    """Tests initialize with an empty artifacts dir."""
    mocked_get_fuzz_targets.return_value = ['fuzz-target']
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'out')
      artifacts_path = os.path.join(out_path, 'artifacts')
      os.makedirs(artifacts_path)
      runner = self._create_runner(workspace=tmp_dir)
      self.assertTrue(runner.initialize())
      mocked_log_error.assert_not_called()
      self.assertTrue(os.path.isdir(artifacts_path))

  @mock.patch('utils.get_fuzz_targets')
  @mock.patch('logging.error')
  def test_initialize_no_artifacts(self, mocked_log_error,
                                   mocked_get_fuzz_targets):
    """Tests initialize with no artifacts dir (the expected setting)."""
    mocked_get_fuzz_targets.return_value = ['fuzz-target']
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'out')
      os.makedirs(out_path)
      runner = self._create_runner(workspace=tmp_dir)
      self.assertTrue(runner.initialize())
      mocked_log_error.assert_not_called()
      self.assertTrue(os.path.isdir(os.path.join(out_path, 'artifacts')))

  def test_initialize_no_fuzz_targets(self):
    """Tests initialize with no fuzz targets."""
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'out')
      os.makedirs(out_path)
      expected_error_args = ('No fuzz targets were found in out directory: %s.',
                             out_path)
      self._test_initialize_fail(expected_error_args, workspace=tmp_dir)

  def test_get_fuzz_target_artifact(self):
    """Tests that get_fuzz_target_artifact works as intended."""
    runner = self._create_runner()
    artifacts_dir = 'artifacts-dir'
    runner.artifacts_dir = artifacts_dir
    artifact_name = 'artifact-name'
    target = mock.MagicMock()
    target_name = 'target_name'
    target.target_name = target_name
    fuzz_target_artifact = runner.get_fuzz_target_artifact(
        target, artifact_name)
    expected_fuzz_target_artifact = (
        'artifacts-dir/target_name-address-artifact-name')
    self.assertEqual(fuzz_target_artifact, expected_fuzz_target_artifact)


class CiFuzzTargetRunnerTest(fake_filesystem_unittest.TestCase):
  """Tests that CiFuzzTargetRunner works as intended."""

  def setUp(self):
    self.setUpPyfakefs()

  @mock.patch('utils.get_fuzz_targets')
  @mock.patch('run_fuzzers.CiFuzzTargetRunner.run_fuzz_target')
  @mock.patch('run_fuzzers.CiFuzzTargetRunner.create_fuzz_target_obj')
  def test_run_fuzz_targets_quits(self, mocked_create_fuzz_target_obj,
                                  mocked_run_fuzz_target,
                                  mocked_get_fuzz_targets):
    """Tests that run_fuzz_targets quits on the first crash it finds."""
    workspace = 'workspace'
    out_path = os.path.join(workspace, 'out')
    self.fs.create_dir(out_path)
    config = _create_config(fuzz_seconds=FUZZ_SECONDS,
                            workspace=workspace,
                            project_name=EXAMPLE_PROJECT)
    runner = run_fuzzers.CiFuzzTargetRunner(config)

    mocked_get_fuzz_targets.return_value = ['target1', 'target2']
    runner.initialize()
    testcase = os.path.join(workspace, 'testcase')
    self.fs.create_file(testcase)
    stacktrace = b'stacktrace'
    mocked_run_fuzz_target.return_value = fuzz_target.FuzzResult(
        testcase, stacktrace)
    magic_mock = mock.MagicMock()
    magic_mock.target_name = 'target1'
    mocked_create_fuzz_target_obj.return_value = magic_mock
    self.assertTrue(runner.run_fuzz_targets())
    self.assertIn('target1-address-testcase', os.listdir(runner.artifacts_dir))
    self.assertEqual(mocked_run_fuzz_target.call_count, 1)


class BatchFuzzTargetRunnerTest(fake_filesystem_unittest.TestCase):
  """Tests that CiFuzzTargetRunner works as intended."""

  def setUp(self):
    self.setUpPyfakefs()

  @mock.patch('utils.get_fuzz_targets')
  @mock.patch('run_fuzzers.BatchFuzzTargetRunner.run_fuzz_target')
  @mock.patch('run_fuzzers.BatchFuzzTargetRunner.create_fuzz_target_obj')
  def test_run_fuzz_targets_quits(self, mocked_create_fuzz_target_obj,
                                  mocked_run_fuzz_target,
                                  mocked_get_fuzz_targets):
    """Tests that run_fuzz_targets doesn't quit on the first crash it finds."""
    workspace = 'workspace'
    out_path = os.path.join(workspace, 'out')
    self.fs.create_dir(out_path)
    config = _create_config(fuzz_seconds=FUZZ_SECONDS,
                            workspace=workspace,
                            project_name=EXAMPLE_PROJECT)
    runner = run_fuzzers.BatchFuzzTargetRunner(config)

    mocked_get_fuzz_targets.return_value = ['target1', 'target2']
    runner.initialize()
    testcase1 = os.path.join(workspace, 'testcase-aaa')
    testcase2 = os.path.join(workspace, 'testcase-bbb')
    self.fs.create_file(testcase1)
    self.fs.create_file(testcase2)
    stacktrace = b'stacktrace'
    call_count = 0

    def mock_run_fuzz_target(_):
      nonlocal call_count
      if call_count == 0:
        testcase = testcase1
      elif call_count == 1:
        testcase = testcase2
      assert call_count != 2
      call_count += 1
      return fuzz_target.FuzzResult(testcase, stacktrace)

    mocked_run_fuzz_target.side_effect = mock_run_fuzz_target
    magic_mock = mock.MagicMock()
    magic_mock.target_name = 'target1'
    mocked_create_fuzz_target_obj.return_value = magic_mock
    self.assertTrue(runner.run_fuzz_targets())
    self.assertIn('target1-address-testcase-aaa',
                  os.listdir(runner.artifacts_dir))
    self.assertEqual(mocked_run_fuzz_target.call_count, 2)


class RunAddressFuzzersIntegrationTest(RunFuzzerIntegrationTestMixin,
                                       unittest.TestCase):
  """Integration tests for build_fuzzers with an ASAN build."""

  BUILD_DIR_NAME = 'cifuzz-latest-build'

  @unittest.skipIf(not os.getenv('INTEGRATION_TESTS'),
                   'INTEGRATION_TESTS=1 not set')
  def test_new_bug_found(self):
    """Tests run_fuzzers with a valid ASAN build."""
    # Set the first return value to True, then the second to False to
    # emulate a bug existing in the current PR but not on the downloaded
    # OSS-Fuzz build.
    with mock.patch('fuzz_target.FuzzTarget.is_reproducible',
                    side_effect=[True, False]):
      with tempfile.TemporaryDirectory() as tmp_dir:
        workspace = os.path.join(tmp_dir, 'workspace')
        shutil.copytree(TEST_DATA_PATH, workspace)
        config = _create_config(fuzz_seconds=FUZZ_SECONDS,
                                workspace=workspace,
                                project_name=EXAMPLE_PROJECT)
        result = run_fuzzers.run_fuzzers(config)
        self.assertEqual(result, run_fuzzers.RunFuzzersResult.BUG_FOUND)
        build_dir = os.path.join(workspace, 'out', self.BUILD_DIR_NAME)
        self.assertNotEqual(0, len(os.listdir(build_dir)))

  @unittest.skipIf(not os.getenv('INTEGRATION_TESTS'),
                   'INTEGRATION_TESTS=1 not set')
  @mock.patch('fuzz_target.FuzzTarget.is_reproducible',
              side_effect=[True, True])
  def test_old_bug_found(self, _):
    """Tests run_fuzzers with a bug found in OSS-Fuzz before."""
    config = _create_config(fuzz_seconds=FUZZ_SECONDS,
                            workspace=TEST_DATA_PATH,
                            project_name=EXAMPLE_PROJECT)
    with tempfile.TemporaryDirectory() as tmp_dir:
      workspace = os.path.join(tmp_dir, 'workspace')
      shutil.copytree(TEST_DATA_PATH, workspace)
      config = _create_config(fuzz_seconds=FUZZ_SECONDS,
                              workspace=TEST_DATA_PATH,
                              project_name=EXAMPLE_PROJECT)
      result = run_fuzzers.run_fuzzers(config)
      self.assertEqual(result, run_fuzzers.RunFuzzersResult.NO_BUG_FOUND)
      build_dir = os.path.join(TEST_DATA_PATH, 'out', self.BUILD_DIR_NAME)
      self.assertTrue(os.path.exists(build_dir))
      self.assertNotEqual(0, len(os.listdir(build_dir)))

  def test_invalid_build(self):
    """Tests run_fuzzers with an invalid ASAN build."""
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'out')
      os.mkdir(out_path)
      config = _create_config(fuzz_seconds=FUZZ_SECONDS,
                              workspace=tmp_dir,
                              project_name=EXAMPLE_PROJECT)
      result = run_fuzzers.run_fuzzers(config)
    self.assertEqual(result, run_fuzzers.RunFuzzersResult.ERROR)


if __name__ == '__main__':
  unittest.main()