| # Copyright 2018 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Script for performing tasks that are useful for fuzzer development. |
| |
| Run "cros_fuzz" in the chroot for a list of command or "cros_fuzz $COMMAND |
| --help" for their full details. Below is a summary of commands that the script |
| can perform: |
| |
| coverage: Generate a coverage report for a given fuzzer (specified by "--fuzzer" |
| option). You almost certainly want to specify the package to build (using |
| the "--package" option) so that a coverage build is done, since a coverage |
| build is needed to generate a report. If your fuzz target is running on |
| ClusterFuzz already, you can use the "--download" option to download the |
| corpus from ClusterFuzz. Otherwise, you can use the "--corpus" option to |
| specify the path of the corpus to run the fuzzer on and generate a report. |
| The corpus will be copied to the sysroot so that the fuzzer can use it. |
| Note that "--download" and "--corpus" are mutually exclusive. |
| |
| reproduce: Runs the fuzzer specified by the "--fuzzer" option on a testcase |
| (path specified by the "--testcase" argument). Optionally does a build when |
| the "--package" option is used. The type of build can be specified using the |
| "--build_type" argument. |
| |
| download: Downloads the corpus from ClusterFuzz of the fuzzer specified by the |
| "--fuzzer" option. The path of the directory the corpus directory is |
| downloaded to can be specified using the "--directory" option. |
| |
| shell: Sets up the sysroot for fuzzing and then chroots into the sysroot giving |
| you a shell that is ready to fuzz. |
| |
| setup: Sets up the sysroot for fuzzing (done prior to doing "reproduce", "shell" |
| and "coverage" commands). |
| |
| cleanup: Undoes "setup". |
| |
| Note that cros_fuzz will print every shell command it runs if you set the |
| log-level to debug ("--log-level debug"). Otherwise, it will print commands that |
| fail. |
| """ |
| |
| import logging |
| import os |
| import shutil |
| |
| from chromite.third_party import lddtree |
| from chromite.third_party.pyelftools.elftools.elf.elffile import ELFFile |
| |
| from chromite.lib import build_target_lib |
| from chromite.lib import commandline |
| from chromite.lib import constants |
| from chromite.lib import cros_build_lib |
| from chromite.lib import gs |
| from chromite.lib import osutils |
| from chromite.lib import portage_util |
| |
| |
| # Directory in sysroot's /tmp directory that this script will use for files it |
| # needs to write. We need a directory to write files to because this script uses |
| # external programs that must write and read to/from files and because these |
| # must be run inside the sysroot and thus are usually unable to read or write |
| # from directories in the chroot environment this script is executed in. |
| SCRIPT_STORAGE_DIRECTORY = "fuzz" |
| SCRIPT_STORAGE_PATH = os.path.join("/", "tmp", SCRIPT_STORAGE_DIRECTORY) |
| |
| # Names of subdirectories in "fuzz" directory used by this script to store |
| # things. |
| CORPUS_DIRECTORY_NAME = "corpus" |
| TESTCASE_DIRECTORY_NAME = "testcase" |
| COVERAGE_REPORT_DIRECTORY_NAME = "coverage-report" |
| |
| # Constants for names of libFuzzer command line options. |
| RUNS_OPTION_NAME = "runs" |
| MAX_TOTAL_TIME_OPTION_NAME = "max_total_time" |
| |
| # The default path a profraw file written by a clang coverage instrumented |
| # binary when run by this script (default is current working directory). |
| DEFAULT_PROFRAW_PATH = "/default.profraw" |
| |
| # Constants for libFuzzer command line values. |
| # 0 runs means execute everything in the corpus and do no mutations. |
| RUNS_DEFAULT_VALUE = 0 |
| # An arbitrary but short amount of time to run a fuzzer to get some coverage |
| # data (when a corpus hasn't been provided and we aren't told to download one). |
| MAX_TOTAL_TIME_DEFAULT_VALUE = 30 |
| |
| |
| class BuildType: |
| """Class to hold the different kinds of build types.""" |
| |
| ASAN = "asan" |
| MSAN = "msan" |
| UBSAN = "ubsan" |
| COVERAGE = "coverage" |
| STANDARD = "" |
| |
| # Build types that users can specify. |
| CHOICES = (ASAN, MSAN, UBSAN, COVERAGE) |
| |
| |
| class SysrootPath: |
| """Class for representing a path that is in the sysroot. |
| |
| Useful for dealing with paths that we must interact with when chrooted into |
| the sysroot and outside of it. |
| |
| For example, if we need to interact with the "/tmp" directory of the |
| sysroot, SysrootPath('/tmp').sysroot returns the path of the directory if we |
| are in chrooted into the sysroot, i.e. "/tmp". |
| |
| SysrootPath('/tmp').chroot returns the path of the directory when in the |
| cros_sdk i.e. SYSROOT_DIRECTORY + "/tmp" (this will probably be |
| "/build/amd64-generic/tmp" in most cases). |
| """ |
| |
| # The actual path to the sysroot (from within the chroot). |
| path_to_sysroot = None |
| |
| def __init__(self, path): |
| """Constructor. |
| |
| Args: |
| path: An absolute path representing something in the sysroot. |
| """ |
| |
| assert path.startswith("/") |
| if self.IsPathInSysroot(path): |
| path = self.FromChrootPathInSysroot(os.path.abspath(path)) |
| self.path_list = path.split(os.sep)[1:] |
| |
| @classmethod |
| def SetPathToSysroot(cls, board): |
| """Sets path_to_sysroot |
| |
| Args: |
| board: The board we will use for our sysroot. |
| |
| Returns: |
| The path to the sysroot (the value of path_to_sysroot). |
| """ |
| cls.path_to_sysroot = build_target_lib.get_default_sysroot_path(board) |
| return cls.path_to_sysroot |
| |
| @property |
| def chroot(self): |
| """Get the path of the object in the Chrome OS SDK chroot. |
| |
| Returns: |
| The path this object represents when chrooted into the sysroot. |
| """ |
| assert ( |
| self.path_to_sysroot is not None |
| ), "set SysrootPath.path_to_sysroot" |
| return os.path.join(self.path_to_sysroot, *self.path_list) |
| |
| @property |
| def sysroot(self): |
| """Get the path of the object when in the sysroot. |
| |
| Returns: |
| The path this object represents when in the Chrome OS SDK . |
| """ |
| return os.path.join("/", *self.path_list) |
| |
| @classmethod |
| def IsPathInSysroot(cls, path): |
| """Is a path in the sysroot. |
| |
| Args: |
| path: The path we are checking is in the sysroot. |
| |
| Returns: |
| True if path is within the sysroot's path in the chroot. |
| """ |
| assert cls.path_to_sysroot |
| return path.startswith(cls.path_to_sysroot) |
| |
| @classmethod |
| def FromChrootPathInSysroot(cls, path): |
| """Converts a chroot-relative path in sysroot into sysroot-relative. |
| |
| Args: |
| path: The chroot-relative path we are converting to sysroot |
| relative. |
| |
| Returns: |
| The sysroot relative version of |path|. |
| """ |
| assert cls.IsPathInSysroot(path) |
| common_prefix = os.path.commonprefix([cls.path_to_sysroot, path]) |
| return path[len(common_prefix) :] |
| |
| |
| def GetScriptStoragePath(relative_path): |
| """Get the SysrootPath representing a script storage path. |
| |
| Get a path of a directory this script will store things in. |
| |
| Args: |
| relative_path: The path relative to the root of the script storage |
| directory. |
| |
| Returns: |
| The SysrootPath representing absolute path of |relative_path| in the |
| script storage directory. |
| """ |
| path = os.path.join(SCRIPT_STORAGE_PATH, relative_path) |
| return SysrootPath(path) |
| |
| |
| def GetSysrootPath(path): |
| """Get the chroot-relative path of a path in the sysroot. |
| |
| Args: |
| path: An absolute path in the sysroot that we will get the path in the |
| chroot for. |
| |
| Returns: |
| The chroot-relative path of |path| in the sysroot. |
| """ |
| return SysrootPath(path).chroot |
| |
| |
| def GetCoverageDirectory(fuzzer): |
| """Get a coverage report directory for a fuzzer |
| |
| Args: |
| fuzzer: The fuzzer to get the coverage report directory for. |
| |
| Returns: |
| The location of the coverage report directory for the |fuzzer|. |
| """ |
| relative_path = os.path.join(COVERAGE_REPORT_DIRECTORY_NAME, fuzzer) |
| return GetScriptStoragePath(relative_path) |
| |
| |
| def GetFuzzerSysrootPath(fuzzer): |
| """Get the path in the sysroot of a fuzzer. |
| |
| Args: |
| fuzzer: The fuzzer to get the path of. |
| |
| Returns: |
| The path of |fuzzer| in the sysroot. |
| """ |
| return SysrootPath(os.path.join("/", "usr", "libexec", "fuzzers", fuzzer)) |
| |
| |
| def GetProfdataPath(fuzzer): |
| """Get the profdata file of a fuzzer. |
| |
| Args: |
| fuzzer: The fuzzer to get the profdata file of. |
| |
| Returns: |
| The path of the profdata file that should be used by |fuzzer|. |
| """ |
| return GetScriptStoragePath("%s.profdata" % fuzzer) |
| |
| |
| def GetPathForCopy(parent_directory, chroot_path): |
| """Returns a path in the script storage directory to copy chroot_path. |
| |
| Returns a SysrootPath representing the location where |chroot_path| should |
| be copied. This path will be in the parent_directory which will be in the |
| script storage directory. |
| """ |
| basename = os.path.basename(chroot_path) |
| return GetScriptStoragePath(os.path.join(parent_directory, basename)) |
| |
| |
| def CopyCorpusToSysroot(src_corpus_path): |
| """Copies corpus into the sysroot. |
| |
| Copies corpus into the sysroot. Doesn't copy if corpus is already in |
| sysroot. |
| |
| Args: |
| src_corpus_path: A path (in the chroot) to a corpus that will be copied |
| into sysroot. |
| |
| Returns: |
| The path in the sysroot that the corpus was copied to. |
| """ |
| if src_corpus_path is None: |
| return None |
| |
| if SysrootPath.IsPathInSysroot(src_corpus_path): |
| # Don't copy if |src_testcase_path| is already in sysroot. Just return |
| # it in the format expected by the caller. |
| return SysrootPath(src_corpus_path) |
| |
| dest_corpus_path = GetPathForCopy(CORPUS_DIRECTORY_NAME, src_corpus_path) |
| osutils.RmDir(dest_corpus_path.chroot, ignore_missing=True) |
| shutil.copytree(src_corpus_path, dest_corpus_path.chroot) |
| return dest_corpus_path |
| |
| |
| def CopyTestcaseToSysroot(src_testcase_path): |
| """Copies a testcase into the sysroot. |
| |
| Copies a testcase into the sysroot. Doesn't copy if testcase is already in |
| sysroot. |
| |
| Args: |
| src_testcase_path: A path (in the chroot) to a testcase that will be |
| copied into sysroot. |
| |
| Returns: |
| The path in the sysroot that the testcase was copied to. |
| """ |
| if SysrootPath.IsPathInSysroot(src_testcase_path): |
| # Don't copy if |src_testcase_path| is already in sysroot. Just return |
| # it in the format expected by the caller. |
| return SysrootPath(src_testcase_path) |
| |
| dest_testcase_path = GetPathForCopy( |
| TESTCASE_DIRECTORY_NAME, src_testcase_path |
| ) |
| osutils.SafeMakedirsNonRoot(os.path.dirname(dest_testcase_path.chroot)) |
| osutils.SafeUnlink(dest_testcase_path.chroot) |
| |
| shutil.copy(src_testcase_path, dest_testcase_path.chroot) |
| return dest_testcase_path |
| |
| |
| def sudo_run(*args, **kwargs): |
| """Wrapper around cros_build_lib.sudo_run. |
| |
| Wrapper that calls cros_build_lib.sudo_run but sets debug_level by |
| default. |
| |
| Args: |
| *args: Positional arguments to pass to cros_build_lib.sudo_run. |
| **kwargs: Keyword arguments to pass to cros_build_lib.sudo_run. |
| |
| Returns: |
| The value returned by calling cros_build_lib.sudo_run. |
| """ |
| kwargs.setdefault("debug_level", logging.DEBUG) |
| return cros_build_lib.sudo_run(*args, **kwargs) |
| |
| |
| def GetLibFuzzerOption(option_name, option_value): |
| """Gets the libFuzzer command line option with the specified name and value. |
| |
| Args: |
| option_name: The name of the libFuzzer option. |
| option_value: The value of the libFuzzer option. |
| |
| Returns: |
| The libFuzzer option composed of |option_name| and |option_value|. |
| """ |
| return "-%s=%s" % (option_name, option_value) |
| |
| |
| def IsOptionLimit(option): |
| """Determines if fuzzer option limits fuzzing time.""" |
| for limit_name in [MAX_TOTAL_TIME_OPTION_NAME, RUNS_OPTION_NAME]: |
| if option.startswith("-%s" % limit_name): |
| return True |
| |
| return False |
| |
| |
| def LimitFuzzing(fuzz_command, corpus): |
| """Limits how long fuzzing will go if unspecified. |
| |
| Adds a reasonable limit on how much fuzzing will be done unless there |
| already is some kind of limit. Mutates fuzz_command. |
| |
| Args: |
| fuzz_command: A command to run a fuzzer. Used to determine if a limit |
| needs to be set. Mutated if it is needed to specify a limit. |
| corpus: The corpus that will be passed to the fuzzer. If not None then |
| fuzzing is limited by running everything in the corpus once. |
| """ |
| if any(IsOptionLimit(x) for x in fuzz_command[1:]): |
| # Don't do anything if there is already a limit. |
| return |
| |
| if corpus: |
| # If there is a corpus, just run everything in the corpus once. |
| fuzz_command.append( |
| GetLibFuzzerOption(RUNS_OPTION_NAME, RUNS_DEFAULT_VALUE) |
| ) |
| return |
| |
| # Since there is no corpus, just fuzz for 30 seconds. |
| logging.info( |
| "Limiting fuzzing to %s seconds.", MAX_TOTAL_TIME_DEFAULT_VALUE |
| ) |
| max_total_time_option = GetLibFuzzerOption( |
| MAX_TOTAL_TIME_OPTION_NAME, MAX_TOTAL_TIME_DEFAULT_VALUE |
| ) |
| fuzz_command.append(max_total_time_option) |
| |
| |
| def GetFuzzExtraEnv(extra_options=None): |
| """Gets extra_env for fuzzing. |
| |
| Gets environment variables and values for running libFuzzer. Sets defaults |
| and allows user to specify extra sanitizer options. |
| |
| Args: |
| extra_options: A dict containing sanitizer options to set in addition to |
| the defaults. |
| |
| Returns: |
| A dict containing environment variables and their values that can be |
| used in the environment libFuzzer runs in. |
| """ |
| if extra_options is None: |
| extra_options = {} |
| |
| # log_path must be set because Chrome OS's patched compiler changes it. |
| # disable odr violation since many fuzzers hit it and it is also disabled on |
| # clusterfuzz. |
| # handle_sigtrap is useful for catching int3 in assertion checks in ChromeOS |
| # code. |
| options_dict = { |
| "log_path": "stderr", |
| "detect_odr_violation": "0", |
| "handle_sigtrap": "1", |
| } |
| options_dict.update(extra_options) |
| sanitizer_options = ":".join("%s=%s" % x for x in options_dict.items()) |
| sanitizers = ("ASAN", "MSAN", "UBSAN") |
| return {x + "_OPTIONS": sanitizer_options for x in sanitizers} |
| |
| |
| def RunFuzzer( |
| fuzzer, |
| corpus_path=None, |
| fuzz_args="", |
| testcase_path=None, |
| crash_expected=False, |
| ): |
| """Runs the fuzzer while chrooted into the sysroot. |
| |
| Args: |
| fuzzer: The fuzzer to run. |
| corpus_path: A path to a corpus (not necessarily in the sysroot) to run |
| the fuzzer on. |
| fuzz_args: Additional arguments to pass to the fuzzer when running it. |
| testcase_path: A path to a testcase (not necessarily in the sysroot) to |
| run the fuzzer on. |
| crash_expected: Is it normal for the fuzzer to crash on this run? |
| """ |
| logging.info("Running fuzzer: %s", fuzzer) |
| fuzzer_sysroot_path = GetFuzzerSysrootPath(fuzzer) |
| fuzz_command = [fuzzer_sysroot_path.sysroot] |
| fuzz_command += fuzz_args.split() |
| |
| if testcase_path: |
| fuzz_command.append(testcase_path) |
| else: |
| LimitFuzzing(fuzz_command, corpus_path) |
| |
| if corpus_path: |
| fuzz_command.append(corpus_path) |
| |
| if crash_expected: |
| # Don't return nonzero when fuzzer OOMs, leaks, or timesout, since we |
| # don't want an exception in those cases. The user may be trying to |
| # reproduce those issues. |
| fuzz_command += ["-error_exitcode=0", "-timeout_exitcode=0"] |
| |
| # We must set exitcode=0 or else the fuzzer will return nonzero on |
| # successful reproduction. |
| sanitizer_options = {"exitcode": "0"} |
| else: |
| sanitizer_options = {} |
| |
| extra_env = GetFuzzExtraEnv(sanitizer_options) |
| RunSysrootCommand( |
| fuzz_command, extra_env=extra_env, debug_level=logging.INFO |
| ) |
| |
| |
| def MergeProfraw(fuzzer): |
| """Merges profraw file from a fuzzer and creates a profdata file. |
| |
| Args: |
| fuzzer: The fuzzer to merge the profraw file from. |
| """ |
| profdata_path = GetProfdataPath(fuzzer) |
| command = [ |
| "llvm-profdata", |
| "merge", |
| "-sparse", |
| DEFAULT_PROFRAW_PATH, |
| "-o", |
| profdata_path.sysroot, |
| ] |
| |
| RunSysrootCommand(command) |
| return profdata_path |
| |
| |
| def GenerateCoverageReport(fuzzer, shared_libraries): |
| """Generates an HTML coverage report from a fuzzer run. |
| |
| Args: |
| fuzzer: The fuzzer to generate the coverage report for. |
| shared_libraries: Libraries loaded dynamically by |fuzzer|. |
| |
| Returns: |
| The path of the coverage report. |
| """ |
| fuzzer_path = GetFuzzerSysrootPath(fuzzer).chroot |
| command = ["llvm-cov", "show", "-object", fuzzer_path] |
| for library in shared_libraries: |
| command += ["-object", library] |
| |
| coverage_directory = GetCoverageDirectory(fuzzer) |
| command += [ |
| "-format=html", |
| "-instr-profile=%s" % GetProfdataPath(fuzzer).chroot, |
| "-output-dir=%s" % coverage_directory.chroot, |
| ] |
| |
| # TODO(metzman): Investigate error messages printed by this command. |
| cros_build_lib.run(command, stderr=True, debug_level=logging.DEBUG) |
| return coverage_directory |
| |
| |
| def GetSharedLibraries(binary_path): |
| """Gets the shared libraries used by a binary. |
| |
| Gets the shared libraries used by the binary. Based on GetSharedLibraries |
| from src/tools/code_coverage/coverage_utils.py in Chromium. |
| |
| Args: |
| binary_path: The path to the binary we want to find the shared libraries |
| of. |
| |
| Returns: |
| The shared libraries used by |binary_path|. |
| """ |
| logging.info("Finding shared libraries for targets (if any).") |
| shared_libraries = [] |
| elf_dict = lddtree.ParseELF( |
| binary_path.chroot, root=SysrootPath.path_to_sysroot |
| ) |
| for shared_library in elf_dict["libs"].values(): |
| shared_library_path = shared_library["path"] |
| |
| if shared_library_path in shared_libraries: |
| continue |
| |
| assert os.path.exists(shared_library_path), ( |
| 'Shared library "%s" used by ' |
| "the given target(s) does not " |
| "exist." % shared_library_path |
| ) |
| |
| if IsInstrumentedWithClangCoverage(shared_library_path): |
| # Do not add non-instrumented libraries. Otherwise, llvm-cov errors |
| # out. |
| shared_libraries.append(shared_library_path) |
| |
| logging.debug( |
| "Found shared libraries (%d): %s.", |
| len(shared_libraries), |
| shared_libraries, |
| ) |
| logging.info("Finished finding shared libraries for targets.") |
| return shared_libraries |
| |
| |
| def IsInstrumentedWithClangCoverage(binary_path): |
| """Determines if a binary is instrumented with clang source based coverage. |
| |
| Args: |
| binary_path: The path of the binary (executable or library) we are |
| checking is instrumented with clang source based coverage. |
| |
| Returns: |
| True if the binary is instrumented with clang source based coverage. |
| """ |
| with open(binary_path, "rb") as file_handle: |
| elf_file = ELFFile(file_handle) |
| return elf_file.get_section_by_name(b"__llvm_covmap") is not None |
| |
| |
| def RunFuzzerAndGenerateCoverageReport(fuzzer, corpus, fuzz_args): |
| """Runs a fuzzer generates a coverage report and returns the report's path. |
| |
| Gets a coverage report for a fuzzer. |
| |
| Args: |
| fuzzer: The fuzzer to run and generate the coverage report for. |
| corpus: The path to a corpus to run the fuzzer on. |
| fuzz_args: Additional arguments to pass to the fuzzer. |
| |
| Returns: |
| The path to the coverage report. |
| """ |
| corpus_path = CopyCorpusToSysroot(corpus) |
| if corpus_path: |
| corpus_path = corpus_path.sysroot |
| |
| RunFuzzer(fuzzer, corpus_path=corpus_path, fuzz_args=fuzz_args) |
| MergeProfraw(fuzzer) |
| fuzzer_sysroot_path = GetFuzzerSysrootPath(fuzzer) |
| shared_libraries = GetSharedLibraries(fuzzer_sysroot_path) |
| return GenerateCoverageReport(fuzzer, shared_libraries) |
| |
| |
| def RunSysrootCommand(command, **kwargs): |
| """Runs command while chrooted into sysroot and returns the output. |
| |
| Args: |
| command: A command to run in the sysroot. |
| **kwargs: Extra arguments to pass to cros_build_lib.sudo_run. |
| |
| Returns: |
| The result of a call to cros_build_lib.sudo_run. |
| """ |
| command = ["chroot", SysrootPath.path_to_sysroot] + command |
| return sudo_run(command, **kwargs) |
| |
| |
| def GetBuildExtraEnv(build_type): |
| """Gets the extra_env for building a package. |
| |
| Args: |
| build_type: The type of build we want to do. |
| |
| Returns: |
| The extra_env to use when building. |
| """ |
| if build_type is None: |
| build_type = BuildType.ASAN |
| |
| use_flags = os.environ.get("USE", "").split() |
| # Check that the user hasn't already set USE flags that we can set. |
| # No good way to iterate over an enum in python2. |
| for use_flag in BuildType.CHOICES: |
| if use_flag in use_flags: |
| logging.warning( |
| "%s in USE flags. Please use --build_type instead.", use_flag |
| ) |
| |
| # Set USE flags. |
| fuzzer_build_type = "fuzzer" |
| use_flags += [fuzzer_build_type, build_type] |
| features_flags = os.environ.get("FEATURES", "").split() |
| if build_type == BuildType.COVERAGE: |
| # We must use ASan when doing coverage builds. |
| use_flags.append(BuildType.ASAN) |
| # Use noclean so that a coverage report can be generated based on the |
| # source code. |
| features_flags.append("noclean") |
| |
| return { |
| "FEATURES": " ".join(features_flags), |
| "USE": " ".join(use_flags), |
| } |
| |
| |
| def BuildPackage(package, board, build_type): |
| """Builds a package on a specified board. |
| |
| Args: |
| package: The package to build. Nothing is built if None. |
| board: The board to build the package on. |
| build_type: The type of the build to do (e.g. asan, msan, ubsan, |
| coverage). |
| """ |
| if package is None: |
| return |
| |
| logging.info("Building %s using %s.", package, build_type) |
| extra_env = GetBuildExtraEnv(build_type) |
| command = [ |
| "cros", |
| "build-packages", |
| "--board", |
| board, |
| "--skip-chroot-upgrade", |
| package, |
| ] |
| # For msan builds, always use "--no-usepkg" since all package needs to be |
| # instrumented with msan. |
| if build_type == BuildType.MSAN: |
| command += ["--no-usepkg"] |
| |
| # Print the output of the build command. Do this because it is familiar to |
| # devs and we don't want to leave them not knowing about the build's |
| # progress for a long time. |
| cros_build_lib.run(command, extra_env=extra_env) |
| |
| |
| def DownloadFuzzerCorpus(fuzzer, dest_directory=None): |
| """Downloads a corpus and returns its path. |
| |
| Downloads a corpus to a subdirectory of dest_directory if specified and |
| returns path on the filesystem of the corpus. Asks users to authenticate |
| if permission to read from bucket is denied. |
| |
| Args: |
| fuzzer: The name of the fuzzer whose corpus we want to download. |
| dest_directory: The directory to download the corpus to. |
| |
| Returns: |
| The path to the downloaded corpus. |
| |
| Raises: |
| gs.NoSuchKey: A corpus for the fuzzer doesn't exist. |
| gs.GSCommandError: The corpus failed to download for another reason. |
| """ |
| if not fuzzer.startswith("chromeos_"): |
| # ClusterFuzz internally appends "chromeos_" to chromeos targets' names. |
| # Therefore we must do so in order to find the corpus. |
| fuzzer = "chromeos_%s" % fuzzer |
| |
| if dest_directory is None: |
| dest_directory = GetScriptStoragePath(CORPUS_DIRECTORY_NAME).chroot |
| osutils.SafeMakedirsNonRoot(dest_directory) |
| |
| clusterfuzz_gcs_corpus_bucket = "chromeos-corpus" |
| suburl = "libfuzzer/%s" % fuzzer |
| gcs_path = gs.GetGsURL( |
| clusterfuzz_gcs_corpus_bucket, |
| for_gsutil=True, |
| public=False, |
| suburl=suburl, |
| ) |
| |
| dest_path = os.path.join(dest_directory, fuzzer) |
| |
| try: |
| logging.info("Downloading corpus to %s.", dest_path) |
| ctx = gs.GSContext() |
| ctx.Copy( |
| gcs_path, |
| dest_directory, |
| recursive=True, |
| parallel=True, |
| debug_level=logging.DEBUG, |
| ) |
| logging.info("Finished downloading corpus.") |
| except gs.GSNoSuchKey as exception: |
| logging.error("Corpus for fuzzer: %s does not exist.", fuzzer) |
| raise exception |
| # Try to authenticate if we were denied permission to access the corpus. |
| except gs.GSCommandError as exception: |
| logging.error( |
| "gsutil failed to download the corpus. You may need to log in. " |
| "See:\n" |
| "https://chromium.googlesource.com/chromiumos/docs/+/HEAD/gsutil.md" |
| "#setup\n" |
| "for instructions on doing this." |
| ) |
| raise exception |
| |
| return dest_path |
| |
| |
| def Reproduce(fuzzer, testcase_path): |
| """Runs a fuzzer in the sysroot on a testcase. |
| |
| Args: |
| fuzzer: The fuzzer to run. |
| testcase_path: The path (not necessarily in the sysroot) of the testcase |
| to run the fuzzer on. |
| """ |
| testcase_sysroot_path = CopyTestcaseToSysroot(testcase_path).sysroot |
| RunFuzzer(fuzzer, testcase_path=testcase_sysroot_path, crash_expected=True) |
| |
| |
| def SetUpSysrootForFuzzing(): |
| """Sets up the sysroot for fuzzing |
| |
| Prepares the sysroot for fuzzing. Idempotent. |
| """ |
| logging.info("Setting up sysroot for fuzzing.") |
| # TODO(metzman): Don't create devices or mount /proc, use platform2_test.py |
| # instead. |
| # Mount /proc in sysroot and setup dev there because they are needed by |
| # sanitizers. |
| proc_manager = ProcManager() |
| proc_manager.Mount() |
| |
| # Setup devices in /dev that are needed by libFuzzer. |
| device_manager = DeviceManager() |
| device_manager.SetUp() |
| |
| # Set up asan_symbolize.py, llvm-symbolizer, and llvm-profdata in the |
| # sysroot so that fuzzer output (including stack traces) can be symbolized |
| # and so that coverage reports can be generated. |
| tool_manager = ToolManager() |
| tool_manager.Install() |
| |
| osutils.SafeMakedirsNonRoot(GetSysrootPath(SCRIPT_STORAGE_PATH)) |
| |
| |
| def CleanUpSysroot(): |
| """Cleans up the the sysroot from SetUpSysrootForFuzzing. |
| |
| Undoes SetUpSysrootForFuzzing. Idempotent. |
| """ |
| logging.info("Cleaning up the sysroot.") |
| proc_manager = ProcManager() |
| proc_manager.Unmount() |
| |
| device_manager = DeviceManager() |
| device_manager.CleanUp() |
| |
| tool_manager = ToolManager() |
| tool_manager.Uninstall() |
| osutils.RmDir(GetSysrootPath(SCRIPT_STORAGE_PATH), ignore_missing=True) |
| |
| |
| class ToolManager: |
| """Class that installs or uninstalls fuzzing tools to/from the sysroot. |
| |
| Install and Uninstall methods are idempotent. Both are safe to call at any |
| point. |
| """ |
| |
| # Path to asan_symbolize.py. |
| ASAN_SYMBOLIZE_PATH = os.path.join("/", "usr", "bin", "asan_symbolize.py") |
| |
| # List of LLVM binaries we must install in sysroot. |
| LLVM_BINARY_NAMES = ["gdbserver", "llvm-symbolizer", "llvm-profdata"] |
| |
| def __init__(self): |
| self.asan_symbolize_sysroot_path = GetSysrootPath( |
| self.ASAN_SYMBOLIZE_PATH |
| ) |
| |
| def Install(self): |
| """Installs tools to the sysroot.""" |
| # Install asan_symbolize.py. |
| sudo_run( |
| ["cp", self.ASAN_SYMBOLIZE_PATH, self.asan_symbolize_sysroot_path] |
| ) |
| # Install the LLVM binaries. |
| # TODO(metzman): Build these tools so that we don't mess up when board |
| # is for a different ISA. |
| for llvm_binary in self._GetLLVMBinaries(): |
| llvm_binary.Install() |
| |
| def Uninstall(self): |
| """Uninstalls tools from the sysroot. Undoes Install.""" |
| # Uninstall asan_symbolize.py. |
| osutils.SafeUnlink(self.asan_symbolize_sysroot_path, sudo=True) |
| # Uninstall the LLVM binaries. |
| for llvm_binary in self._GetLLVMBinaries(): |
| llvm_binary.Uninstall() |
| |
| def _GetLLVMBinaries(self): |
| """Creates LlvmBinary objects for each binary in LLVM_BINARY_NAMES.""" |
| return [LlvmBinary(x) for x in self.LLVM_BINARY_NAMES] |
| |
| |
| class LlvmBinary: |
| """Class for representing installing/uninstalling an LLVM binary in sysroot. |
| |
| Install and Uninstall methods are idempotent. Both are safe to call at any |
| time. |
| """ |
| |
| # Path to the lddtree chromite script. |
| LDDTREE_SCRIPT_PATH = constants.CHROMITE_BIN_DIR / "lddtree" |
| |
| def __init__(self, binary): |
| self.binary = binary |
| self.install_dir = GetSysrootPath( |
| os.path.join("/", "usr", "libexec", binary) |
| ) |
| self.binary_dir_path = GetSysrootPath(os.path.join("/", "usr", "bin")) |
| self.binary_chroot_dest_path = os.path.join( |
| self.binary_dir_path, binary |
| ) |
| |
| def Uninstall(self): |
| """Removes an LLVM binary from sysroot. Undoes Install.""" |
| osutils.RmDir(self.install_dir, ignore_missing=True, sudo=True) |
| osutils.SafeUnlink(self.binary_chroot_dest_path, sudo=True) |
| |
| def Install(self): |
| """Installs (sets up) an LLVM binary in the sysroot. |
| |
| Sets up an llvm binary in the sysroot so that it can be run there. |
| """ |
| # Create a directory for installing |binary| and all of its dependencies |
| # in the sysroot. |
| binary_rel_path = ["usr", "bin", self.binary] |
| binary_chroot_path = os.path.join("/", *binary_rel_path) |
| if not os.path.exists(binary_chroot_path): |
| logging.warning( |
| "Cannot copy %s, file does not exist in chroot.", |
| binary_chroot_path, |
| ) |
| logging.warning( |
| "Functionality provided by %s will be missing.", |
| binary_chroot_path, |
| ) |
| return |
| |
| osutils.SafeMakedirsNonRoot(self.install_dir) |
| |
| # Copy the binary and everything needed to run it into the sysroot. |
| cmd = [ |
| self.LDDTREE_SCRIPT_PATH, |
| "-v", |
| "--generate-wrappers", |
| "--root", |
| "/", |
| "--copy-to-tree", |
| self.install_dir, |
| binary_chroot_path, |
| ] |
| sudo_run(cmd) |
| |
| # Create a symlink to the copy of the binary (we can't do lddtree in |
| # self.binary_dir_path). Note that symlink should be relative so that it |
| # will be valid when chrooted into the sysroot. |
| rel_path = os.path.relpath(self.install_dir, self.binary_dir_path) |
| link_path = os.path.join(rel_path, *binary_rel_path) |
| osutils.SafeSymlink(link_path, self.binary_chroot_dest_path, sudo=True) |
| |
| |
| class DeviceManager: |
| """Class that creates or removes devices from /dev in sysroot. |
| |
| SetUp and CleanUp methods are idempotent. Both are safe to call at any |
| point. |
| """ |
| |
| DEVICE_MKNOD_PARAMS = { |
| "null": (666, 3), |
| "random": (444, 8), |
| "urandom": (444, 9), |
| } |
| |
| MKNOD_MAJOR = "1" |
| |
| def __init__(self): |
| self.dev_path_chroot = GetSysrootPath("/dev") |
| |
| def _GetDevicePath(self, device_name): |
| """Returns the path of |device_name| in sysroot's /dev.""" |
| return os.path.join(self.dev_path_chroot, device_name) |
| |
| def SetUp(self): |
| """Sets up devices in the sysroot's /dev. |
| |
| Creates /dev/null, /dev/random, and /dev/urandom. If they already exist |
| then recreates them. |
| """ |
| self.CleanUp() |
| osutils.SafeMakedirsNonRoot(self.dev_path_chroot) |
| for device, mknod_params in self.DEVICE_MKNOD_PARAMS.items(): |
| device_path = self._GetDevicePath(device) |
| self._MakeCharDevice(device_path, *mknod_params) |
| |
| def CleanUp(self): |
| """Cleans up devices in the sysroot's /dev. Undoes SetUp. |
| |
| Removes /dev/null, /dev/random, and /dev/urandom if they exist. |
| """ |
| for device in self.DEVICE_MKNOD_PARAMS: |
| device_path = self._GetDevicePath(device) |
| if os.path.exists(device_path): |
| # Use -r since dev/null is sometimes a directory. |
| sudo_run(["rm", "-r", device_path]) |
| |
| def _MakeCharDevice(self, path, mode, minor): |
| """Make a character device.""" |
| mode = str(mode) |
| minor = str(minor) |
| command = ["mknod", "-m", mode, path, "c", self.MKNOD_MAJOR, minor] |
| sudo_run(command) |
| |
| |
| class ProcManager: |
| """Class that mounts or unmounts /proc in sysroot. |
| |
| Mount and Unmount are idempotent. Both are safe to call at any point. |
| """ |
| |
| PROC_PATH = "/proc" |
| |
| def __init__(self): |
| self.proc_path_chroot = GetSysrootPath(self.PROC_PATH) |
| self.is_mounted = osutils.IsMounted(self.proc_path_chroot) |
| |
| def Unmount(self): |
| """Unmounts /proc in chroot. Undoes Mount.""" |
| if not self.is_mounted: |
| return |
| osutils.UmountDir(self.proc_path_chroot, cleanup=False) |
| |
| def Mount(self): |
| """Mounts /proc in chroot. Remounts it if already mounted.""" |
| self.Unmount() |
| osutils.MountDir( |
| self.PROC_PATH, |
| self.proc_path_chroot, |
| "proc", |
| debug_level=logging.DEBUG, |
| ) |
| |
| |
| def EnterSysrootShell(): |
| """Spawns and gives user access to a bash shell in the sysroot.""" |
| command = ["/bin/bash", "-i"] |
| return RunSysrootCommand( |
| command, |
| extra_env=GetFuzzExtraEnv(), |
| debug_level=logging.INFO, |
| check=False, |
| ).returncode |
| |
| |
| def StripFuzzerPrefixes(fuzzer_name): |
| """Strip the prefix ClusterFuzz uses in case they are specified. |
| |
| Strip the prefixes used by ClusterFuzz if the users has included them by |
| accident. |
| |
| Args: |
| fuzzer_name: The fuzzer whose name may contain prefixes. |
| |
| Returns: |
| The name of the fuzz target without prefixes. |
| """ |
| initial_name = fuzzer_name |
| |
| def StripPrefix(prefix): |
| if fuzzer_name.startswith(prefix): |
| return fuzzer_name[len(prefix) :] |
| return fuzzer_name |
| |
| clusterfuzz_prefixes = ["libFuzzer_", "chromeos_"] |
| |
| for prefix in clusterfuzz_prefixes: |
| fuzzer_name = StripPrefix(prefix) |
| |
| if initial_name != fuzzer_name: |
| logging.warning( |
| "%s contains a prefix from ClusterFuzz (one or more of %s) that is " |
| "not part of the fuzzer's name. Interpreting --fuzzer as %s.", |
| initial_name, |
| clusterfuzz_prefixes, |
| fuzzer_name, |
| ) |
| |
| return fuzzer_name |
| |
| |
| def ExecuteShellCommand(): |
| """Executes the "shell" command. |
| |
| Sets up the sysroot for fuzzing and gives user access to a bash shell it |
| spawns in the sysroot. |
| |
| Returns: |
| The exit code of the shell command. |
| """ |
| SetUpSysrootForFuzzing() |
| return EnterSysrootShell() |
| |
| |
| def ExecuteSetupCommand(): |
| """Executes the "setup" command. Wrapper for SetUpSysrootForFuzzing. |
| |
| Sets up the sysroot for fuzzing. |
| """ |
| SetUpSysrootForFuzzing() |
| |
| |
| def ExecuteCleanupCommand(): |
| """Executes the "cleanup" command. Wrapper for CleanUpSysroot. |
| |
| Undoes pre-fuzzing setup. |
| """ |
| CleanUpSysroot() |
| |
| |
| def ExecuteCoverageCommand(options): |
| """Executes the "coverage" command. |
| |
| Executes the "coverage" command by optionally doing a coverage build of a |
| package, optionally downloading the fuzzer's corpus, optionally copying it |
| into the sysroot, running the fuzzer and then generating a coverage report |
| for the user to view. Causes program to exit if fuzzer is not instrumented |
| with source based coverage. |
| |
| Args: |
| options: The parsed arguments passed to this program. |
| """ |
| BuildPackage(options.package, options.board, BuildType.COVERAGE) |
| |
| fuzzer = StripFuzzerPrefixes(options.fuzzer) |
| fuzzer_sysroot_path = GetFuzzerSysrootPath(fuzzer) |
| if not IsInstrumentedWithClangCoverage(fuzzer_sysroot_path.chroot): |
| # Don't run the fuzzer if it isn't instrumented with source based |
| # coverage. Quit and let the user know how to build the fuzzer properly. |
| cros_build_lib.Die( |
| "%s is not instrumented with source based coverage.\nSpecify " |
| "--package to do a coverage build or build with USE flag: " |
| '"coverage".', |
| fuzzer, |
| ) |
| |
| corpus = options.corpus |
| if options.download: |
| corpus = DownloadFuzzerCorpus(options.fuzzer) |
| |
| # Set up sysroot for fuzzing. |
| SetUpSysrootForFuzzing() |
| |
| coverage_report_path = RunFuzzerAndGenerateCoverageReport( |
| fuzzer, corpus, options.fuzz_args |
| ) |
| |
| # Get path on host so user can access it with their browser. |
| # TODO(metzman): Add the ability to convert to host paths to path_util. |
| external_trunk_path = os.getenv("EXTERNAL_TRUNK_PATH") |
| coverage_report_host_path = os.path.join( |
| external_trunk_path, "chroot", coverage_report_path.chroot[1:] |
| ) |
| print( |
| "Coverage report written to file://%s/index.html" |
| % coverage_report_host_path |
| ) |
| |
| |
| def ExecuteDownloadCommand(options): |
| """Executes the "download" command. Wrapper around DownloadFuzzerCorpus.""" |
| DownloadFuzzerCorpus(StripFuzzerPrefixes(options.fuzzer), options.directory) |
| |
| |
| def ExecuteReproduceCommand(options): |
| """Executes the "reproduce" command. |
| |
| Executes the "reproduce" command by Running a fuzzer on a testcase. |
| May build the fuzzer before running. |
| |
| Args: |
| options: The parsed arguments passed to this program. |
| """ |
| if options.build_type and not options.package: |
| raise Exception( |
| "Cannot specify --build_type without specifying --package." |
| ) |
| |
| # Verify that "msan-fuzzer" profile is being used with msan. |
| # Check presence of "-fsanitize=memory" in CFLAGS. |
| if options.build_type == BuildType.MSAN: |
| cmd = ["portageq-%s" % options.board, "envvar", "CFLAGS"] |
| cflags = cros_build_lib.run( |
| cmd, capture_output=True, encoding="utf-8" |
| ).stdout.splitlines() |
| check_string = "-fsanitize=memory" |
| if not any(check_string in s for s in cflags): |
| logging.error( |
| "-fsanitize=memory not found in CFLAGS. " |
| 'Use "setup_board --board=amd64-generic --profile=msan-fuzzer" ' |
| "for MSan Fuzzing Builds." |
| ) |
| raise Exception("Incompatible profile used for msan fuzzing.") |
| |
| BuildPackage(options.package, options.board, options.build_type) |
| SetUpSysrootForFuzzing() |
| Reproduce(StripFuzzerPrefixes(options.fuzzer), options.testcase) |
| |
| |
| def InstallBaseDependencies(options): |
| """Installs the base packages needed to chroot in board sysroot. |
| |
| Args: |
| options: The parsed arguments passed to this program. |
| """ |
| package = "virtual/implicit-system" |
| if not portage_util.IsPackageInstalled( |
| package, sysroot=SysrootPath.path_to_sysroot |
| ): |
| build_type = getattr(options, "build_type", None) |
| BuildPackage(package, options.board, build_type) |
| |
| |
| def ParseArgs(argv): |
| """Parses program arguments. |
| |
| Args: |
| argv: The program arguments we want to parse. |
| |
| Returns: |
| An options object which will tell us which command to run and which |
| options to use for that command. |
| """ |
| parser = commandline.ArgumentParser(description=__doc__) |
| |
| parser.add_argument( |
| "--board", |
| default=cros_build_lib.GetDefaultBoard(), |
| help="Board on which to run test.", |
| ) |
| |
| subparsers = parser.add_subparsers(dest="command") |
| |
| subparsers.add_parser("cleanup", help="Undo setup command.") |
| coverage_parser = subparsers.add_parser( |
| "coverage", help="Get a coverage report for a fuzzer." |
| ) |
| |
| coverage_parser.add_argument("--package", help="Package to build.") |
| |
| corpus_parser = coverage_parser.add_mutually_exclusive_group() |
| corpus_parser.add_argument("--corpus", help="Corpus to run fuzzer on.") |
| |
| corpus_parser.add_argument( |
| "--download", |
| action="store_true", |
| help="Generate coverage report based on corpus from ClusterFuzz.", |
| ) |
| |
| coverage_parser.add_argument( |
| "--fuzzer", |
| required=True, |
| help="The fuzz target to generate a coverage report for.", |
| ) |
| |
| coverage_parser.add_argument( |
| "--fuzz-args", |
| default="", |
| help="Arguments to pass libFuzzer. " |
| "Please use an equals sign or parsing will fail " |
| '(i.e. --fuzzer_args="-rss_limit_mb=2048 -print_funcs=1").', |
| ) |
| |
| download_parser = subparsers.add_parser( |
| "download", help="Download a corpus." |
| ) |
| |
| download_parser.add_argument( |
| "--directory", help="Path to directory to download the corpus to." |
| ) |
| |
| download_parser.add_argument( |
| "--fuzzer", required=True, help="Fuzzer to download the corpus for." |
| ) |
| |
| reproduce_parser = subparsers.add_parser( |
| "reproduce", help="Run a fuzzer on a testcase." |
| ) |
| |
| reproduce_parser.add_argument( |
| "--testcase", required=True, help="Path of testcase to run fuzzer on." |
| ) |
| |
| reproduce_parser.add_argument( |
| "--fuzzer", required=True, help="Fuzzer to reproduce the crash on." |
| ) |
| |
| reproduce_parser.add_argument("--package", help="Package to build.") |
| |
| reproduce_parser.add_argument( |
| "--build-type", |
| choices=BuildType.CHOICES, |
| help="Type of build.", |
| type=str.lower, |
| ) # Ignore sanitizer case. |
| |
| subparsers.add_parser("setup", help="Set up the sysroot to test fuzzing.") |
| |
| subparsers.add_parser( |
| "shell", |
| help="Set up sysroot for fuzzing and get a shell in the sysroot.", |
| ) |
| |
| opts = parser.parse_args(argv) |
| opts.Freeze() |
| return opts |
| |
| |
| def main(argv): |
| """Parses arguments and executes a command. |
| |
| Args: |
| argv: The program arguments. |
| |
| Returns: |
| 0 on success. Non-zero on failure. |
| """ |
| cros_build_lib.AssertInsideChroot() |
| options = ParseArgs(argv) |
| if options.board is None: |
| logging.error('Please specify "--board" or set ".default_board".') |
| return 1 |
| |
| SysrootPath.SetPathToSysroot(options.board) |
| |
| InstallBaseDependencies(options) |
| |
| if options.command == "cleanup": |
| ExecuteCleanupCommand() |
| elif options.command == "coverage": |
| ExecuteCoverageCommand(options) |
| elif options.command == "setup": |
| ExecuteSetupCommand() |
| elif options.command == "download": |
| ExecuteDownloadCommand(options) |
| elif options.command == "reproduce": |
| ExecuteReproduceCommand(options) |
| elif options.command == "shell": |
| return ExecuteShellCommand() |
| |
| return 0 |