1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
|
#!/usr/bin/env python
# Copyright 2018 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Run a single fuzz target built with code coverage instrumentation.
This script assumes that corresponding corpus was downloaded via gclient sync
and saved to: src/testing/libfuzzer/fuzzer_corpus/{fuzzer_name}/.
"""
import argparse
import glob
import json
import logging
import os
import shutil
import signal
import subprocess
import sys
import time
import zipfile
_THIS_DIR = os.path.dirname(os.path.realpath(__file__))
# Path to the fuzzer corpus directory that is used for bots.
_CORPUS_FOR_BOTS_DIR = os.path.join(_THIS_DIR, os.path.pardir, os.path.pardir,
'testing', 'libfuzzer',
'fuzzer_corpus_for_bots')
# Dummy corpus in case real corpus doesn't exist.
_DUMMY_INPUT_CONTENTS = 'dummy input just to have at least one corpus unit'
_DUMMY_INPUT_FILENAME = 'dummy_corpus_input'
# Used for running fuzzer targets in code coverage config.
_DUMMY_CORPUS_DIRECTORY = 'dummy_corpus_dir_which_should_be_empty'
_LIBFUZZER_FLAGS = ['-merge=1', '-timeout=60', '-rss_limit_mb=8192']
_SLEEP_DURATION_SECONDS = 8
def _PrepareCorpus(fuzzer_name, output_dir):
"""Prepares the corpus to run fuzzer target.
If a corpus for bots is available, use it directly, otherwise, creates a
dummy corpus.
Args:
fuzzer_name (str): Name of the fuzzer to create corpus for.
output_dir (str): An output directory to store artifacts.
Returns:
A path to the directory of the prepared corpus.
"""
corpus_dir = os.path.join(output_dir, fuzzer_name + '_corpus')
_RecreateDir(corpus_dir)
corpus_for_bots = glob.glob(
os.path.join(os.path.abspath(_CORPUS_FOR_BOTS_DIR), fuzzer_name, '*.zip'))
if len(corpus_for_bots) >= 2:
raise RuntimeError(
'Expected only one, but multiple versions of corpus exit')
if len(corpus_for_bots) == 1:
zipfile.ZipFile(corpus_for_bots[0]).extractall(path=corpus_dir)
return corpus_dir
logging.info('Corpus for %s does not exist, create a dummy corpus input',
fuzzer_name)
dummy_input_path = os.path.join(corpus_dir, _DUMMY_INPUT_FILENAME)
with open(dummy_input_path, 'wb') as fh:
fh.write(_DUMMY_INPUT_CONTENTS)
return corpus_dir
def _ParseCommandArguments():
"""Adds and parses relevant arguments for tool comands.
Returns:
A dictionary representing the arguments.
"""
arg_parser = argparse.ArgumentParser()
arg_parser.add_argument(
'-f',
'--fuzzer',
type=str,
required=True,
help='Path to the fuzz target executable.')
arg_parser.add_argument(
'-o',
'--output-dir',
type=str,
required=True,
help='Output directory where corpus and coverage dumps can be stored in.')
arg_parser.add_argument(
'-t',
'--timeout',
type=int,
required=True,
help='Timeout value for running a single fuzz target.')
# Ignored. Used to comply with isolated script contract, see chromium_tests
# and swarming recipe modules for more details.
arg_parser.add_argument(
'--isolated-script-test-output',
type=str,
required=False,
help=argparse.SUPPRESS)
# Ditto.
arg_parser.add_argument(
'--isolated-script-test-perf-output',
type=str,
required=False,
help=argparse.SUPPRESS)
if len(sys.argv) == 1:
arg_parser.print_help()
sys.exit(1)
args = arg_parser.parse_args()
assert os.path.isfile(
args.fuzzer), ("Fuzzer '%s' does not exist." % args.fuzzer)
assert os.path.isdir(
args.output_dir), ("Output dir '%s' does not exist." % args.output_dir)
assert args.timeout > 0, 'Invalid timeout value: %d.' % args.timeout
return args
def _RecreateDir(dir_path):
if os.path.exists(dir_path):
shutil.rmtree(dir_path)
os.mkdir(dir_path)
def _RunFuzzTarget(fuzzer, fuzzer_name, output_dir, corpus_dir, timeout):
# The way we run fuzz targets in code coverage config (-merge=1) requires an
# empty directory to be provided to fuzz target. We run fuzz targets with
# -merge=1 because that mode is crash-resistant.
dummy_corpus_dir = os.path.join(output_dir, _DUMMY_CORPUS_DIRECTORY)
_RecreateDir(dummy_corpus_dir)
cmd = [fuzzer] + _LIBFUZZER_FLAGS + [dummy_corpus_dir, corpus_dir]
try:
_RunWithTimeout(cmd, timeout)
except Exception as e:
logging.info('Failed to run %s: %s', fuzzer_name, e)
shutil.rmtree(dummy_corpus_dir)
def _RunWithTimeout(cmd, timeout):
logging.info('Run fuzz target using the following command: %s', str(cmd))
# TODO: we may need to use |creationflags=subprocess.CREATE_NEW_PROCESS_GROUP|
# on Windows or send |signal.CTRL_C_EVENT| signal if the process times out.
runner = subprocess.Popen(cmd)
timer = 0
while timer < timeout and runner.poll() is None:
time.sleep(_SLEEP_DURATION_SECONDS)
timer += _SLEEP_DURATION_SECONDS
if runner.poll() is None:
try:
logging.info('Fuzz target timed out, interrupting it.')
# libFuzzer may spawn some child processes, that is why we have to call
# os.killpg, which would send the signal to our Python process as well, so
# we just catch and ignore it in this try block.
os.killpg(os.getpgid(runner.pid), signal.SIGINT)
except KeyboardInterrupt:
# Python's default signal handler raises KeyboardInterrupt exception for
# SIGINT, suppress it here to prevent interrupting the script itself.
pass
runner.communicate()
logging.info('Finished running the fuzz target.')
def Main():
log_format = '[%(asctime)s %(levelname)s] %(message)s'
logging.basicConfig(level=logging.INFO, format=log_format)
args = _ParseCommandArguments()
fuzzer_name = os.path.splitext(os.path.basename(args.fuzzer))[0]
corpus_dir = _PrepareCorpus(fuzzer_name, args.output_dir)
start_time = time.time()
_RunFuzzTarget(args.fuzzer, fuzzer_name, args.output_dir, corpus_dir,
args.timeout)
end_time = time.time()
shutil.rmtree(corpus_dir)
if args.isolated_script_test_output:
# TODO(crbug.com/913827): Actually comply with the isolated script contract
# on src/testing/scripts/common.
with open(args.isolated_script_test_output, 'w') as f:
json.dump({
'version': 3,
'interrupted': False,
'path_delimiter': '.',
'seconds_since_epoch': int(start_time),
'num_failures_by_type': {
'FAIL': 0,
'PASS': 1
},
'num_regressions': 0,
'tests': {
fuzzer_name: {
'expected': 'PASS',
'actual': 'PASS',
'times': [int(end_time - start_time),]
},
}
}, f)
return 0
if __name__ == '__main__':
sys.exit(Main())
|
