| #!/usr/bin/env python2.7 |
| # Copyright 2019 The Fuchsia Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import os |
| import unittest |
| |
| import test_env |
| from test_case import TestCaseWithFuzzer |
| |
| |
| class FuzzerTest(TestCaseWithFuzzer): |
| |
| def test_is_resolved(self): |
| # The default package is automatically resolved; use another. |
| self.create_fuzzer('fake-package2/fake-target1', resolve=False) |
| |
| cmd = ['pkgctl', 'pkg-status', self.fuzzer.package_url] |
| self.assertFalse(self.fuzzer.is_resolved()) |
| self.assertSsh(*cmd) |
| |
| self.resolve_fuzzer() |
| self.assertTrue(self.fuzzer.is_resolved()) |
| self.assertSsh(*cmd) |
| |
| def test_resolve(self): |
| # The default package is automatically resolved; use another. |
| self.create_fuzzer('fake-package2/fake-target1', resolve=False) |
| status_cmd = ['pkgctl', 'pkg-status', self.fuzzer.package_url] |
| resolve_cmd = ['pkgctl', 'resolve', self.fuzzer.package_url] |
| |
| # Resolution failure |
| self.assertError( |
| lambda: self.fuzzer.resolve(), |
| 'Failed to resolve package: fake-package2') |
| self.assertSsh(*status_cmd) |
| self.assertSsh(*resolve_cmd) |
| |
| # This is a little tricky: to test this, the first call to |
| # 'pkgctl pkg-status' should fail, but the second, after a call to |
| # 'pkgctl resolve', should succeed. |
| package_path = '/pkgfs/versions/deadbeef' |
| |
| self.set_outputs( |
| status_cmd, ['Package on disk: no'], end=10.0, ssh=True, returncode=2) |
| self.set_outputs( |
| status_cmd, ['Package on disk: yes (path={})'.format(package_path)], |
| start=10.0, |
| reset=False, |
| ssh=True) |
| |
| process = self.get_process(resolve_cmd, ssh=True) |
| process.duration = 10.0 |
| |
| self.fuzzer.resolve() |
| self.assertSsh(*status_cmd) |
| self.assertSsh(*resolve_cmd) |
| self.assertEqual(self.fuzzer.package_path, package_path) |
| |
| def test_list_artifacts(self): |
| artifacts = ['crash-deadbeef', 'leak-deadfa11', 'oom-feedface'] |
| artifacts = [os.path.join(self.fuzzer.output, a) for a in artifacts] |
| for artifact in artifacts: |
| self.host.touch(artifact) |
| self.assertEqual(self.fuzzer.list_artifacts(), artifacts) |
| |
| def test_is_running(self): |
| fuzzer1 = self.create_fuzzer('check', 'fake-package1/fake-target1') |
| fuzzer2 = self.create_fuzzer('check', 'fake-package1/fake-target2') |
| |
| self.assertFalse(fuzzer1.is_running()) |
| self.assertFalse(fuzzer2.is_running()) |
| |
| self.set_running(fuzzer1.package, fuzzer1.executable, duration=5) |
| self.assertTrue(fuzzer1.is_running()) |
| self.assertFalse(fuzzer2.is_running()) |
| |
| self.set_running(fuzzer2.package, fuzzer2.executable, duration=5) |
| self.assertTrue(fuzzer1.is_running()) |
| self.assertTrue(fuzzer2.is_running()) |
| |
| # Results are cached until refresh |
| self.host.sleep(5) |
| self.assertTrue(fuzzer1.is_running()) |
| self.assertTrue(fuzzer2.is_running()) |
| |
| self.assertFalse(fuzzer1.is_running(refresh=True)) |
| self.assertFalse(fuzzer2.is_running(refresh=True)) |
| |
| def test_require_stopped(self): |
| self.fuzzer.require_stopped() |
| |
| self.set_running( |
| self.fuzzer.package, self.fuzzer.executable, duration=10) |
| self.assertError( |
| lambda: self.fuzzer.require_stopped(), |
| 'fake-package1/fake-target1 is running and must be stopped first.') |
| self.host.sleep(10) |
| self.fuzzer.require_stopped() |
| |
| def test_start(self): |
| self.fuzzer.start() |
| self.assertSsh( |
| 'run', |
| self.fuzzer.executable_url, |
| '-artifact_prefix=data/', |
| '-jobs=1', |
| 'data/corpus', |
| ) |
| |
| def test_start_with_dictionary(self): |
| self.touch_on_device(self.ns.resource_abspath('dictionary')) |
| self.fuzzer.start() |
| self.assertSsh( |
| 'run', |
| self.fuzzer.executable_url, |
| '-artifact_prefix=data/', |
| '-dict=pkg/data/fake-target1/dictionary', |
| '-jobs=1', |
| 'data/corpus', |
| ) |
| |
| def test_start_with_seed_corpus(self): |
| self.touch_on_device(self.ns.resource_abspath('corpus/deadbeef')) |
| self.touch_on_device(self.ns.resource_abspath('corpus/feedface')) |
| self.fuzzer.start() |
| self.assertSsh( |
| 'run', |
| self.fuzzer.executable_url, |
| '-artifact_prefix=data/', |
| '-jobs=1', |
| 'data/corpus', |
| 'pkg/data/fake-target1/corpus', |
| ) |
| |
| def test_start_already_running(self): |
| self.create_fuzzer('start', 'fake-package1/fake-target1') |
| self.set_running(self.fuzzer.package, self.fuzzer.executable) |
| self.assertError( |
| lambda: self.fuzzer.start(), |
| 'fake-package1/fake-target1 is running and must be stopped first.') |
| |
| def test_start_foreground(self): |
| self.create_fuzzer('start', str(self.fuzzer), '--foreground') |
| self.fuzzer.start() |
| self.assertSsh( |
| 'run', |
| self.fuzzer.executable_url, |
| '-artifact_prefix=data/', |
| 'data/corpus', |
| ) |
| |
| def test_start_debug(self): |
| self.create_fuzzer('start', str(self.fuzzer), '--debug') |
| self.fuzzer.start() |
| self.assertSsh( |
| 'run', |
| self.fuzzer.executable_url, |
| '-artifact_prefix=data/', |
| '-handle_abrt=0', |
| '-handle_bus=0', |
| '-handle_fpe=0', |
| '-handle_ill=0', |
| '-handle_segv=0', |
| '-jobs=1', |
| 'data/corpus', |
| ) |
| |
| def test_start_with_options(self): |
| self.create_fuzzer( |
| 'start', str(self.fuzzer), '-option2=foo', '-option1=\'bar baz\'') |
| self.fuzzer.start() |
| self.assertSsh( |
| 'run', |
| self.fuzzer.executable_url, |
| '-artifact_prefix=data/', |
| '-jobs=1', |
| '-option1=\'bar baz\'', |
| '-option2=foo', |
| 'data/corpus', |
| ) |
| |
| def test_start_with_passthrough(self): |
| self.create_fuzzer('start', str(self.fuzzer), '--', '-option1=bar') |
| self.fuzzer.start() |
| self.assertSsh( |
| 'run', |
| self.fuzzer.executable_url, |
| '-artifact_prefix=data/', |
| '-jobs=1', |
| 'data/corpus', |
| '--', |
| '-option1=bar', |
| ) |
| |
| def test_start_failure(self): |
| # Make the fuzzer fail after 15 "seconds". |
| cmd = [ |
| 'run', |
| self.fuzzer.executable_url, |
| '-artifact_prefix=data/', |
| '-jobs=1', |
| 'data/corpus', |
| ] |
| process = self.get_process(cmd, ssh=True) |
| process.duration = 15 |
| process.succeeds = False |
| |
| # Make the log appear after 20 "seconds". |
| self.create_log(start=20) |
| |
| # The fuzzer should fail before the log appears. |
| self.assertError( |
| lambda: self.fuzzer.start(), |
| 'fake-package1/fake-target1 failed to start.') |
| self.assertSsh(*cmd) |
| self.assertEqual(self.host.elapsed, 15) |
| |
| def test_start_slow(self): |
| # Make the log file appear after 15 "seconds". |
| self.create_log(start=15) |
| |
| # Make the fuzzer fail after 20 "seconds". |
| cmd = [ |
| 'run', |
| self.fuzzer.executable_url, |
| '-artifact_prefix=data/', |
| '-jobs=1', |
| 'data/corpus', |
| ] |
| process = self.get_process(cmd, ssh=True) |
| process.duration = 20 |
| process.succeeds = False |
| |
| # The log should appear before the fuzzer fails. |
| self.fuzzer.start() |
| self.assertSsh(*cmd) |
| self.assertEqual(self.host.elapsed, 15) |
| |
| # Helper to test Fuzzer.symbolize with different logs. |
| MUTATION_SEQUENCE = 'MS: 1 SomeMutation; base unit: foo' |
| SYMBOLIZER_OUTPUT = '[[symbolizer output]]' |
| |
| def symbolize_helper(self, inputs, job_num=0, echo=False): |
| # Prime the sumbolization commnad |
| has_mututation_seq = FuzzerTest.MUTATION_SEQUENCE in inputs |
| if has_mututation_seq: |
| cmd = self.symbolize_cmd() |
| self.set_outputs(cmd, [FuzzerTest.SYMBOLIZER_OUTPUT]) |
| |
| # Ensure the output directory exists, but the log file doesn't |
| self.host.mkdir(self.fuzzer._output) |
| self.assertFalse( |
| self.host.isfile( |
| os.path.join(self.fuzzer.output, 'fuzz-latest.log'))) |
| |
| # Create the input log and symbolize it |
| with self.host.open('unsymbolized', 'w+') as unsymbolized: |
| for line in inputs: |
| unsymbolized.write(line) |
| unsymbolized.write('\n') |
| unsymbolized.flush() |
| unsymbolized.seek(0) |
| self.assertEqual( |
| has_mututation_seq, |
| self.fuzzer.symbolize_log(unsymbolized, job_num, echo=echo)) |
| |
| # The log file and a symlink should be created. |
| self.assertTrue(self.host.isfile(self.fuzzer.logfile(job_num))) |
| self.assertTrue( |
| self.host.isfile( |
| os.path.join(self.fuzzer.output, 'fuzz-latest.log'))) |
| |
| # Symbolizer output is only inserted if a mutation sequence was present |
| if has_mututation_seq: |
| self.assertRan(*cmd) |
| i = inputs.index(FuzzerTest.MUTATION_SEQUENCE) |
| outputs = list(inputs[:i]) |
| outputs.append(FuzzerTest.SYMBOLIZER_OUTPUT) |
| outputs += list(inputs[i:]) |
| else: |
| outputs = list(inputs) |
| with self.host.open(self.fuzzer.logfile(job_num)) as symbolized: |
| self.assertEqual(symbolized.read().strip().split('\n'), outputs) |
| |
| def test_symbolize_log_no_mutation_sequence(self): |
| self.symbolize_helper([ |
| 'A line', |
| 'Another line', |
| 'Yet another line', |
| ]) |
| |
| def test_symbolize_log_no_process_id(self): |
| self.symbolize_helper( |
| [ |
| 'A line', |
| 'Another line', |
| FuzzerTest.MUTATION_SEQUENCE, |
| 'Yet another line', |
| 'artifact_prefix=\'data/\'; Test unit written to data/crash-aaaa', |
| ]) |
| self.assertScpFrom( |
| self.ns.data_abspath('crash-aaaa'), self.fuzzer.output) |
| |
| def test_symbolize_log_pid_from_stacktrace(self): |
| self.symbolize_helper( |
| [ |
| 'A line', |
| 'Another line', |
| '==12345== INFO: libfuzzer stack trace:', |
| FuzzerTest.MUTATION_SEQUENCE, |
| 'Yet another line', |
| 'artifact_prefix=\'data/\'; Test unit written to data/crash-bbbb', |
| ]) |
| self.assertScpFrom( |
| self.ns.data_abspath('crash-bbbb'), self.fuzzer.output) |
| |
| def test_symbolize_log_pid_from_deadly_signal(self): |
| self.symbolize_helper( |
| [ |
| 'A line', |
| 'Another line', |
| '==67890== libFuzzer: deadly signal', |
| FuzzerTest.MUTATION_SEQUENCE, |
| 'Yet another line', |
| 'artifact_prefix=\'data/\'; Test unit written to data/crash-cccc', |
| ]) |
| self.assertScpFrom( |
| self.ns.data_abspath('crash-cccc'), self.fuzzer.output) |
| |
| def test_symbolize_log_with_job_num(self): |
| self.symbolize_helper( |
| [ |
| '==67890== libFuzzer: deadly signal', |
| FuzzerTest.MUTATION_SEQUENCE, |
| ], |
| job_num=13) |
| |
| def test_symbolize_log_with_echo(self): |
| self.symbolize_helper( |
| [ |
| '==67890== libFuzzer: deadly signal', |
| FuzzerTest.MUTATION_SEQUENCE, |
| ], |
| echo=True) |
| |
| self.assertLogged( |
| '==67890== libFuzzer: deadly signal', |
| FuzzerTest.SYMBOLIZER_OUTPUT, |
| FuzzerTest.MUTATION_SEQUENCE, |
| ) |
| |
| def test_monitor(self): |
| self.create_fuzzer('start', 'fake-package1/fake-target2') |
| self.set_running( |
| self.fuzzer.package, self.fuzzer.executable, duration=15) |
| |
| # Make the file that scp grabs |
| self.host.mkdir(self.fuzzer.output) |
| |
| logname = os.path.join(self.fuzzer.output, 'fuzz-0.log') |
| with self.host.open(logname, 'w') as log: |
| log.write('==67890== libFuzzer: deadly signal\n') |
| log.write('MS: 1 SomeMutation; base unit: foo\n') |
| |
| # Make another log file to simulate the results of a previous run |
| old_log = os.path.join(self.fuzzer.output, 'fuzz-1234-56-78-9012-0.log') |
| self.host.touch(old_log) |
| |
| # Monitor the fuzzer until it exits |
| self.fuzzer.monitor() |
| self.assertGreaterEqual(self.host.elapsed, 15) |
| |
| # Verify we grabbed the logs and symbolized them. |
| self.assertScpFrom( |
| self.ns.data_abspath('fuzz-[0-9].log'), self.fuzzer.output) |
| cmd = ['rm', '-f', self.ns.data_abspath('fuzz-[0-9].log')] |
| self.assertSsh(*cmd) |
| |
| cmd = self.symbolize_cmd() |
| self.assertRan(*cmd) |
| |
| # Log from scp should have been deleted |
| self.assertFalse(self.host.isfile(logname)) |
| |
| # Old log should be untouched |
| self.assertTrue(self.host.isfile(old_log)) |
| self.assertFalse(self.host.isfile(self.fuzzer.logfile(1))) |
| |
| # New log and symlink should exist |
| self.assertTrue(self.host.isfile(self.fuzzer.logfile(0))) |
| self.assertTrue( |
| self.host.isfile( |
| os.path.join(self.fuzzer.output, 'fuzz-latest.log'))) |
| |
| def test_stop(self): |
| # Stopping when stopped has no effect |
| self.create_fuzzer('stop', 'fake-package1/fake-target2') |
| self.assertFalse(self.fuzzer.is_running()) |
| self.fuzzer.stop() |
| self.assertFalse(self.fuzzer.is_running()) |
| |
| self.set_running(self.fuzzer.package, self.fuzzer.executable) |
| pid = self.device.getpid(self.fuzzer.package, self.fuzzer.executable) |
| self.fuzzer.stop() |
| self.assertSsh('kill', str(pid)) |
| |
| def test_repro(self): |
| # Globs should work, but only if they match files |
| self.create_fuzzer( |
| 'repro', |
| 'package1/target2', |
| 'crash-*', |
| 'oom-feedface', |
| ) |
| |
| # Invalid units |
| self.assertError( |
| lambda: self.fuzzer.repro(), |
| 'No matching files: "crash-* oom-feedface".') |
| |
| # Valid units, but already running |
| self.host.touch('crash-deadbeef') |
| self.host.touch('crash-deadfa11') |
| self.host.touch('oom-feedface') |
| self.set_running( |
| self.fuzzer.package, self.fuzzer.executable, duration=60) |
| self.assertError( |
| lambda: self.fuzzer.repro(), |
| 'fake-package1/fake-target2 is running and must be stopped first.') |
| self.host.sleep(60) |
| |
| # Valid |
| self.fuzzer.repro() |
| self.assertSsh( |
| 'run', |
| self.fuzzer.executable_url, |
| '-artifact_prefix=data/', |
| 'data/crash-deadbeef', |
| 'data/crash-deadfa11', |
| 'data/oom-feedface', |
| ) |
| |
| def test_analyze(self): |
| self.set_running( |
| self.fuzzer.package, self.fuzzer.executable, duration=10) |
| with self.assertRaises(SystemExit): |
| self.fuzzer.analyze() |
| self.host.sleep(10) |
| |
| # Make the log file appear right away. 'touch_on_device' is a bit of a |
| # misleading name here, all it really does is prepare a canned response |
| # to an 'ls' command over SSH. This uses the wildcarded log pattern, |
| # since that is what is passed to `ls` when the fuzzer is starting. |
| self.touch_on_device(self.ns.data_abspath('fuzz-[0-9].log')) |
| |
| # Make the fuzzer run for 60 "seconds". |
| cmd = [ |
| 'run', |
| self.fuzzer.executable_url, |
| '-artifact_prefix=data/', |
| '-jobs=1', |
| '-max_total_time=60', |
| '-print_coverage=1', |
| 'data/corpus', |
| ] |
| process = self.get_process(cmd, ssh=True) |
| process.duration = 60 |
| |
| self.fuzzer.analyze() |
| self.assertSsh(*cmd) |
| |
| # Round to the nearest microsecond |
| self.assertEqual(round(self.host.elapsed, 6), 70) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
