#!/usr/bin/env python # # Copyright (C) 2017 The Android Open Source Project # # 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. # import logging import os import shutil from vts.runners.host import asserts from vts.runners.host import base_test from vts.runners.host import const from vts.runners.host import keys from vts.runners.host import test_runner from vts.utils.python.controllers import adb from vts.utils.python.controllers import android_device from vts.utils.python.common import list_utils from vts.testcases.fuzz.template.libfuzzer_test import libfuzzer_test_config as config from vts.testcases.fuzz.template.libfuzzer_test.libfuzzer_test_case import LibFuzzerTestCase class LibFuzzerTest(base_test.BaseTestClass): """Runs LLVM libfuzzer tests on target. Attributes: _dut: AndroidDevice, the device under test as config. """ def setUpClass(self): """Creates a remote shell instance, and copies data files.""" required_params = [ keys.ConfigKeys.IKEY_DATA_FILE_PATH, keys.ConfigKeys.IKEY_BINARY_TEST_SOURCE, ] self.getUserParams(required_params) logging.info('%s: %s', keys.ConfigKeys.IKEY_DATA_FILE_PATH, self.data_file_path) logging.info('%s: %s', keys.ConfigKeys.IKEY_BINARY_TEST_SOURCE, self.binary_test_source) self._dut = self.android_devices[0] self._dut.stop() self._dut.adb.shell('mkdir %s -p' % config.FUZZER_TEST_DIR) def tearDownClass(self): """Deletes all copied data.""" self._dut.adb.shell('rm -rf %s' % config.FUZZER_TEST_DIR) self._dut.start() def PushFiles(self, src): """adb pushes test case file to target.""" push_src = os.path.join(self.data_file_path, src) push_dst = config.FUZZER_TEST_DIR self._dut.adb.push('%s %s' % (push_src, push_dst), no_except=True) logging.info('Adb pushed: %s \nto: %s', push_src, push_dst) return push_dst def CreateTestCases(self): """Creates LibFuzzerTestCase instances. Returns: LibFuzzerTestCase list. """ test_cases = map( lambda x: LibFuzzerTestCase(x, config.FUZZER_DEFAULT_PARAMS, {}), self.binary_test_source) return test_cases def CreateCorpusOut(self, test_case): """Creates corpus output directory on the target. Args: test_case: LibFuzzerTestCase object, current test case. Throws: throws an AdbError when there is an error in adb operations. """ corpus_out = test_case.GetCorpusOutDir() self._dut.adb.shell('mkdir %s -p' % corpus_out) def RetrieveCorpusSeed(self, test_case): """Retrieves corpus seed directory from GCS to the target. Args: test_case: LibFuzzerTestCase object, current test case. Throws: throws an AdbError when there is an error in adb operations. Returns: inuse_seed, the file path of the inuse seed in GCS, if fetch succeeded. None, otherwise. """ inuse_seed = self._corpus_manager.FetchCorpusSeed( test_case._test_name, self._temp_dir) local_corpus_seed_dir = os.path.join( self._temp_dir, '%s_corpus_seed' % test_case._test_name) if os.path.exists(local_corpus_seed_dir) and os.listdir( local_corpus_seed_dir): self._dut.adb.push(local_corpus_seed_dir, config.FUZZER_TEST_DIR) else: corpus_seed = test_case.GetCorpusSeedDir() self._dut.adb.shell('mkdir %s -p' % corpus_seed) return inuse_seed def AnalyzeGeneratedCorpus(self, test_case): """Analyzes the generated corpus body. Args: test_case: LibFuzzerTestCase object. Returns: number of newly generated corpus strings, if the out directory exists. 0, otherwise. """ logging.info('temporary directory for this test: %s', self._temp_dir) pulled_corpus_out_dir = os.path.join( self._temp_dir, os.path.basename(test_case.GetCorpusOutDir())) if os.path.exists(pulled_corpus_out_dir): logging.info('corpus out directory pulled from target: %s', pulled_corpus_out_dir) pulled_corpus = os.listdir(pulled_corpus_out_dir) logging.debug(pulled_corpus) logging.info('generated corpus size: %d', len(pulled_corpus)) return len(pulled_corpus) else: logging.error('corput out directory does not exist on the host.') return 0 def EvaluateTestcase(self, test_case, result, inuse_seed): """Evaluates the test result and moves the used seed accordingly. Args: test_case: LibFuzzerTestCase object. result: a result dict object returned by the adb shell command. inuse_seed: the seed used as input to this test case. Raises: signals.TestFailure when the testcase failed. """ return_codes = result.get('return_codes', None) if return_codes == config.ExitCode.FUZZER_TEST_PASS: logging.info( 'adb shell fuzzing command exited normally with exitcode %d.', result['return_codes']) if inuse_seed is not None: self._corpus_manager.InuseToDest(test_case._test_name, inuse_seed, 'corpus_complete') elif return_codes == config.ExitCode.FUZZER_TEST_FAIL: logging.info( 'adb shell fuzzing command exited normally with exitcode %d.', result['return_codes']) if inuse_seed is not None: self._corpus_manager.InuseToDest(test_case._test_name, inuse_seed, 'corpus_crash') else: logging.error('adb shell fuzzing command exited abnormally.') if inuse_seed is not None: self._corpus_manager.InuseToDest(test_case._test_name, inuse_seed, 'corpus_error') def RunTestcase(self, test_case): """Runs the given test case and asserts the result. Args: test_case: LibFuzzerTestCase object. """ self.PushFiles(test_case.bin_host_path) self.CreateCorpusOut(test_case) inuse_seed = self.RetrieveCorpusSeed(test_case) if inuse_seed == 'locked': # skip this test case logging.warning('test case locked, skipping testcase %s.', test_case.test_name) return fuzz_cmd = '"%s"' % test_case.GetRunCommand() result = {} try: result = self._dut.adb.shell(fuzz_cmd, no_except=True) except adb.AdbError as e: logging.exception(e) corpus_trigger_dir = os.path.join(self._temp_dir, test_case.GetCorpusTriggerDir()) os.makedirs(corpus_trigger_dir) try: self._dut.adb.pull(config.FUZZER_TEST_CRASH_REPORT, corpus_trigger_dir) except adb.AdbError as e: logging.exception(e) logging.error('crash report was not created during test run.') try: self._dut.adb.pull(test_case.GetCorpusOutDir(), self._temp_dir) self.AnalyzeGeneratedCorpus(test_case) self._corpus_manager.UploadCorpusOutDir(test_case._test_name, self._temp_dir) except adb.AdbError as e: logging.exception(e) logging.error('Device failed. Removing lock from GCS.') self._corpus_manager.remove_lock(test_case._test_name) if inuse_seed is not 'directory': self.EvaluateTestcase(test_case, result, inuse_seed) self.AssertTestResult(test_case, result) def LogCrashReport(self, test_case): """Logs crash-causing fuzzer input. Reads the crash report file and logs the contents in format: '\x01\x23\x45\x67\x89\xab\xcd\xef' Args: test_case: LibFuzzerTestCase object """ touch_cmd = 'touch %s' % config.FUZZER_TEST_CRASH_REPORT self._dut.adb.shell(touch_cmd) # output is string of a hexdump from crash report file. # From the example above, output would be '0123456789abcdef'. xxd_cmd = 'xxd -p %s' % config.FUZZER_TEST_CRASH_REPORT output = self._dut.adb.shell(xxd_cmd) remove_chars = ['\r', '\t', '\n', ' '] for char in remove_chars: output = output.replace(char, '') crash_report = '' # output is guaranteed to be even in length since its a hexdump. for offset in xrange(0, len(output), 2): crash_report += '\\x%s' % output[offset:offset + 2] logging.info('FUZZER_TEST_CRASH_REPORT for %s: "%s"', test_case.test_name, crash_report) # TODO(trong): differentiate between crashes and sanitizer rule violations. def AssertTestResult(self, test_case, result): """Asserts that test case finished as expected. Checks that device is in responsive state. If not, waits for boot then reports test as failure. If it is, asserts that all test commands returned exit code 0. Args: test_case: LibFuzzerTestCase object result: dict(str, str, int), command results from shell. """ logging.info('Test case results.') logging.info('stdout: %s' % result[const.STDOUT]) logging.info('stderr: %s' % result[const.STDERR]) logging.info('exit code: %s' % result[const.EXIT_CODE]) if not self._dut.hasBooted(): self._dut.waitForBootCompletion() asserts.fail('%s left the device in unresponsive state.' % test_case.test_name) exit_code = result[const.EXIT_CODE] if exit_code == config.ExitCode.FUZZER_TEST_FAIL: #TODO(b/64123979): once normal fail happens, examine. self.LogCrashReport(test_case) asserts.fail('%s failed normally.' % test_case.test_name) elif exit_code != config.ExitCode.FUZZER_TEST_PASS: asserts.fail('%s failed abnormally.' % test_case.test_name) def tearDownClass(self): """Removes the temporary directory used for corpus management.""" logging.debug('Temporary directory %s is being deleted', self._temp_dir) try: shutil.rmtree(self._temp_dir) except OSError as e: logging.exception(e) def generateFuzzerTests(self): """Runs fuzzer tests.""" self.runGeneratedTests( test_func=self.RunTestcase, settings=self.CreateTestCases(), name_func=lambda x: x.test_name) if __name__ == '__main__': test_runner.main()