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
|
#!/usr/bin/python3
# (C) 2017,2018 Jack Lloyd
import sys
import os
import subprocess
import optparse # pylint: disable=deprecated-module
import stat
import multiprocessing
import time
def run_fuzzer_gdb(args):
(fuzzer_bin, corpus_file) = args
gdb_proc = subprocess.Popen(['gdb', '--quiet', '--return-child-result', fuzzer_bin],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
close_fds=True)
gdb_commands = ('run < %s\nbt\nquit\n' % (corpus_file)).encode('ascii')
(stdout, stderr) = gdb_proc.communicate(gdb_commands)
if gdb_proc.returncode == 0:
return (0, '', '')
return (corpus_file, gdb_proc.returncode, stdout.decode('ascii'), stderr.decode('ascii'))
def run_fuzzer(args):
(fuzzer_bin, corpus_file) = args
corpus_fd = open(corpus_file, 'r')
fuzzer_proc = subprocess.Popen([fuzzer_bin], stdin=corpus_fd,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True)
(stdout, stderr) = fuzzer_proc.communicate()
corpus_fd.close()
return (corpus_file, fuzzer_proc.returncode, stdout.decode('ascii'), stderr.decode('ascii'))
def run_fuzzer_many_files(fuzzer_bin, corpus_files):
fuzzer_proc = subprocess.Popen([fuzzer_bin] + corpus_files, stdin=None,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True)
(stdout, stderr) = fuzzer_proc.communicate()
return (fuzzer_proc.returncode, stdout.decode('ascii'), stderr.decode('ascii'))
def main(args=None):
#pylint: disable=too-many-branches
#pylint: disable=too-many-statements
#pylint: disable=too-many-locals
if args is None:
args = sys.argv
parser = optparse.OptionParser(
usage='Usage: %prog [options] corpus_dir fuzzers_dir',
)
parser.add_option('--gdb', action='store_true',
help='Run under GDB and capture backtraces')
parser.add_option('--one-at-a-time', action='store_true', default=False,
help='Test one corpus input at a time')
(options, args) = parser.parse_args(args)
if len(args) != 3:
parser.print_usage()
return 1
if options.gdb and not options.one_at_a_time:
print("Option --gdb requires --one-at-a-time")
return 1
corpus_dir = args[1]
fuzzer_dir = args[2]
if not os.access(corpus_dir, os.R_OK):
print("Error could not access corpus directory '%s'" % (corpus_dir))
return 1
if not os.access(fuzzer_dir, os.R_OK):
print("Error could not access fuzzers directory '%s'" % (fuzzer_dir))
return 1
fuzzers = set([])
for fuzzer in os.listdir(fuzzer_dir):
if fuzzer.endswith('.zip'):
continue
fuzzers.add(fuzzer)
corpii = set([])
for corpus in os.listdir(corpus_dir):
# Ignore regular files in toplevel dir
if not stat.S_ISDIR(os.stat(os.path.join(corpus_dir, corpus)).st_mode):
continue
if corpus == '.git':
continue
corpii.add(corpus)
fuzzers_without_corpus = fuzzers - corpii
corpus_without_fuzzers = corpii - fuzzers
for f in sorted(list(fuzzers_without_corpus)):
print("Warning: Fuzzer %s has no corpus" % (f))
for c in sorted(list(corpus_without_fuzzers)):
print("Warning: Corpus %s has no fuzzer" % (c))
fuzzers_with_corpus = fuzzers & corpii
crash_count = 0
stderr_count = 0
stdout_count = 0
if options.one_at_a_time:
pool = multiprocessing.Pool(multiprocessing.cpu_count() * 2)
chunk_size = 32 # arbitrary
run_fuzzer_func = run_fuzzer_gdb if options.gdb else run_fuzzer
for fuzzer in sorted(list(fuzzers_with_corpus)):
fuzzer_bin = os.path.join(fuzzer_dir, fuzzer)
corpus_subdir = os.path.join(corpus_dir, fuzzer)
corpus_files = [os.path.join(corpus_subdir, l) for l in sorted(list(os.listdir(corpus_subdir)))]
# We have to do this hack because multiprocessing's Pool.map doesn't support
# passing any initial arguments, just the single iteratable
map_args = [(fuzzer_bin, f) for f in corpus_files]
start = time.time()
for result in pool.map(run_fuzzer_func, map_args, chunk_size):
(corpus_file, retcode, stdout, stderr) = result
if retcode != 0:
print("Fuzzer %s crashed with input %s returncode %d" % (fuzzer, corpus_file, retcode))
crash_count += 1
if stdout:
print("Fuzzer %s produced stdout on input %s:\n%s" % (fuzzer, corpus_file, stdout))
stdout_count += 1
if stderr:
print("Fuzzer %s produced stderr on input %s:\n%s" % (fuzzer, corpus_file, stderr))
stderr_count += 1
duration = time.time() - start
print("Tested fuzzer %s with %d test cases, %d crashes in %.02f seconds" % (
fuzzer, len(corpus_files), crash_count, duration))
crash_count = 0
sys.stdout.flush()
else:
for fuzzer in sorted(list(fuzzers_with_corpus)):
fuzzer_bin = os.path.join(fuzzer_dir, fuzzer)
corpus_subdir = os.path.join(corpus_dir, fuzzer)
corpus_files = [os.path.join(corpus_subdir, l) for l in sorted(list(os.listdir(corpus_subdir)))]
start = time.time()
(retcode, stdout, stderr) = run_fuzzer_many_files(fuzzer_bin, corpus_files)
if retcode != 0:
print("Fuzzer %s crashed returncode %d" % (fuzzer, retcode))
crash_count += 1
if stdout:
print("Fuzzer %s produced stdout:\n%s" % (fuzzer, stdout))
stdout_count += 1
if stderr:
print("Fuzzer %s produced stderr:\n%s" % (fuzzer, stderr))
stderr_count += 1
duration = time.time() - start
print("Tested fuzzer %s with %d test cases, %d crashes in %.02f seconds" % (
fuzzer, len(corpus_files), crash_count, duration))
crash_count = 0
if crash_count > 0 or stderr_count > 0 or stdout_count > 0:
print("Ran fuzzer tests, %d crashes %d stdout %d stderr" % (crash_count, stdout_count, stderr_count))
return 2
return 0
if __name__ == '__main__':
sys.exit(main())
|
