350 lines
13 KiB
Python
Executable file
350 lines
13 KiB
Python
Executable file
#!/usr/bin/env python
|
|
|
|
"""
|
|
This is a generic fuzz testing tool, see --help for more information.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import random
|
|
import subprocess
|
|
import itertools
|
|
|
|
class TestGenerator:
|
|
def __init__(self, inputs, delete, insert, replace,
|
|
insert_strings, pick_input):
|
|
self.inputs = [(s, open(s).read()) for s in inputs]
|
|
|
|
self.delete = bool(delete)
|
|
self.insert = bool(insert)
|
|
self.replace = bool(replace)
|
|
self.pick_input = bool(pick_input)
|
|
self.insert_strings = list(insert_strings)
|
|
|
|
self.num_positions = sum([len(d) for _,d in self.inputs])
|
|
self.num_insert_strings = len(insert_strings)
|
|
self.num_tests = ((delete + (insert + replace)*self.num_insert_strings)
|
|
* self.num_positions)
|
|
self.num_tests += 1
|
|
|
|
if self.pick_input:
|
|
self.num_tests *= self.num_positions
|
|
|
|
def position_to_source_index(self, position):
|
|
for i,(s,d) in enumerate(self.inputs):
|
|
n = len(d)
|
|
if position < n:
|
|
return (i,position)
|
|
position -= n
|
|
raise ValueError,'Invalid position.'
|
|
|
|
def get_test(self, index):
|
|
assert 0 <= index < self.num_tests
|
|
|
|
picked_position = None
|
|
if self.pick_input:
|
|
index,picked_position = divmod(index, self.num_positions)
|
|
picked_position = self.position_to_source_index(picked_position)
|
|
|
|
if index == 0:
|
|
return ('nothing', None, None, picked_position)
|
|
|
|
index -= 1
|
|
index,position = divmod(index, self.num_positions)
|
|
position = self.position_to_source_index(position)
|
|
if self.delete:
|
|
if index == 0:
|
|
return ('delete', position, None, picked_position)
|
|
index -= 1
|
|
|
|
index,insert_index = divmod(index, self.num_insert_strings)
|
|
insert_str = self.insert_strings[insert_index]
|
|
if self.insert:
|
|
if index == 0:
|
|
return ('insert', position, insert_str, picked_position)
|
|
index -= 1
|
|
|
|
assert self.replace
|
|
assert index == 0
|
|
return ('replace', position, insert_str, picked_position)
|
|
|
|
class TestApplication:
|
|
def __init__(self, tg, test):
|
|
self.tg = tg
|
|
self.test = test
|
|
|
|
def apply(self):
|
|
if self.test[0] == 'nothing':
|
|
pass
|
|
else:
|
|
i,j = self.test[1]
|
|
name,data = self.tg.inputs[i]
|
|
if self.test[0] == 'delete':
|
|
data = data[:j] + data[j+1:]
|
|
elif self.test[0] == 'insert':
|
|
data = data[:j] + self.test[2] + data[j:]
|
|
elif self.test[0] == 'replace':
|
|
data = data[:j] + self.test[2] + data[j+1:]
|
|
else:
|
|
raise ValueError,'Invalid test %r' % self.test
|
|
open(name,'wb').write(data)
|
|
|
|
def revert(self):
|
|
if self.test[0] != 'nothing':
|
|
i,j = self.test[1]
|
|
name,data = self.tg.inputs[i]
|
|
open(name,'wb').write(data)
|
|
|
|
def quote(str):
|
|
return '"' + str + '"'
|
|
|
|
def run_one_test(test_application, index, input_files, args):
|
|
test = test_application.test
|
|
|
|
# Interpolate arguments.
|
|
options = { 'index' : index,
|
|
'inputs' : ' '.join(quote(f) for f in input_files) }
|
|
|
|
# Add picked input interpolation arguments, if used.
|
|
if test[3] is not None:
|
|
pos = test[3][1]
|
|
options['picked_input'] = input_files[test[3][0]]
|
|
options['picked_input_pos'] = pos
|
|
# Compute the line and column.
|
|
file_data = test_application.tg.inputs[test[3][0]][1]
|
|
line = column = 1
|
|
for i in range(pos):
|
|
c = file_data[i]
|
|
if c == '\n':
|
|
line += 1
|
|
column = 1
|
|
else:
|
|
column += 1
|
|
options['picked_input_line'] = line
|
|
options['picked_input_col'] = column
|
|
|
|
test_args = [a % options for a in args]
|
|
if opts.verbose:
|
|
print '%s: note: executing %r' % (sys.argv[0], test_args)
|
|
|
|
stdout = None
|
|
stderr = None
|
|
if opts.log_dir:
|
|
stdout_log_path = os.path.join(opts.log_dir, '%s.out' % index)
|
|
stderr_log_path = os.path.join(opts.log_dir, '%s.err' % index)
|
|
stdout = open(stdout_log_path, 'wb')
|
|
stderr = open(stderr_log_path, 'wb')
|
|
else:
|
|
sys.stdout.flush()
|
|
p = subprocess.Popen(test_args, stdout=stdout, stderr=stderr)
|
|
p.communicate()
|
|
exit_code = p.wait()
|
|
|
|
test_result = (exit_code == opts.expected_exit_code or
|
|
exit_code in opts.extra_exit_codes)
|
|
|
|
if stdout is not None:
|
|
stdout.close()
|
|
stderr.close()
|
|
|
|
# Remove the logs for passes, unless logging all results.
|
|
if not opts.log_all and test_result:
|
|
os.remove(stdout_log_path)
|
|
os.remove(stderr_log_path)
|
|
|
|
if not test_result:
|
|
print 'FAIL: %d' % index
|
|
elif not opts.succinct:
|
|
print 'PASS: %d' % index
|
|
return test_result
|
|
|
|
def main():
|
|
global opts
|
|
from optparse import OptionParser, OptionGroup
|
|
parser = OptionParser("""%prog [options] ... test command args ...
|
|
|
|
%prog is a tool for fuzzing inputs and testing them.
|
|
|
|
The most basic usage is something like:
|
|
|
|
$ %prog --file foo.txt ./test.sh
|
|
|
|
which will run a default list of fuzzing strategies on the input. For each
|
|
fuzzed input, it will overwrite the input files (in place), run the test script,
|
|
then restore the files back to their original contents.
|
|
|
|
NOTE: You should make sure you have a backup copy of your inputs, in case
|
|
something goes wrong!!!
|
|
|
|
You can cause the fuzzing to not restore the original files with
|
|
'--no-revert'. Generally this is used with '--test <index>' to run one failing
|
|
test and then leave the fuzzed inputs in place to examine the failure.
|
|
|
|
For each fuzzed input, %prog will run the test command given on the command
|
|
line. Each argument in the command is subject to string interpolation before
|
|
being executed. The syntax is "%(VARIABLE)FORMAT" where FORMAT is a standard
|
|
printf format, and VARIABLE is one of:
|
|
|
|
'index' - the test index being run
|
|
'inputs' - the full list of test inputs
|
|
'picked_input' - (with --pick-input) the selected input file
|
|
'picked_input_pos' - (with --pick-input) the selected input position
|
|
'picked_input_line' - (with --pick-input) the selected input line
|
|
'picked_input_col' - (with --pick-input) the selected input column
|
|
|
|
By default, the script will run forever continually picking new tests to
|
|
run. You can limit the number of tests that are run with '--max-tests <number>',
|
|
and you can run a particular test with '--test <index>'.
|
|
|
|
You can specify '--stop-on-fail' to stop the script on the first failure
|
|
without reverting the changes.
|
|
|
|
""")
|
|
parser.add_option("-v", "--verbose", help="Show more output",
|
|
action='store_true', dest="verbose", default=False)
|
|
parser.add_option("-s", "--succinct", help="Reduce amount of output",
|
|
action="store_true", dest="succinct", default=False)
|
|
|
|
group = OptionGroup(parser, "Test Execution")
|
|
group.add_option("", "--expected-exit-code", help="Set expected exit code",
|
|
type=int, dest="expected_exit_code",
|
|
default=0)
|
|
group.add_option("", "--extra-exit-code",
|
|
help="Set additional expected exit code",
|
|
type=int, action="append", dest="extra_exit_codes",
|
|
default=[])
|
|
group.add_option("", "--log-dir",
|
|
help="Capture test logs to an output directory",
|
|
type=str, dest="log_dir",
|
|
default=None)
|
|
group.add_option("", "--log-all",
|
|
help="Log all outputs (not just failures)",
|
|
action="store_true", dest="log_all", default=False)
|
|
parser.add_option_group(group)
|
|
|
|
group = OptionGroup(parser, "Input Files")
|
|
group.add_option("", "--file", metavar="PATH",
|
|
help="Add an input file to fuzz",
|
|
type=str, action="append", dest="input_files", default=[])
|
|
group.add_option("", "--filelist", metavar="LIST",
|
|
help="Add a list of inputs files to fuzz (one per line)",
|
|
type=str, action="append", dest="filelists", default=[])
|
|
parser.add_option_group(group)
|
|
|
|
group = OptionGroup(parser, "Fuzz Options")
|
|
group.add_option("", "--replacement-chars", dest="replacement_chars",
|
|
help="Characters to insert/replace",
|
|
default="0{}[]<>\;@#$^%& ")
|
|
group.add_option("", "--replacement-string", dest="replacement_strings",
|
|
action="append", help="Add a replacement string to use",
|
|
default=[])
|
|
group.add_option("", "--replacement-list", dest="replacement_lists",
|
|
help="Add a list of replacement strings (one per line)",
|
|
action="append", default=[])
|
|
group.add_option("", "--no-delete", help="Don't delete characters",
|
|
action='store_false', dest="enable_delete", default=True)
|
|
group.add_option("", "--no-insert", help="Don't insert strings",
|
|
action='store_false', dest="enable_insert", default=True)
|
|
group.add_option("", "--no-replace", help="Don't replace strings",
|
|
action='store_false', dest="enable_replace", default=True)
|
|
group.add_option("", "--no-revert", help="Don't revert changes",
|
|
action='store_false', dest="revert", default=True)
|
|
group.add_option("", "--stop-on-fail", help="Stop on first failure",
|
|
action='store_true', dest="stop_on_fail", default=False)
|
|
parser.add_option_group(group)
|
|
|
|
group = OptionGroup(parser, "Test Selection")
|
|
group.add_option("", "--test", help="Run a particular test",
|
|
type=int, dest="test", default=None, metavar="INDEX")
|
|
group.add_option("", "--max-tests", help="Maximum number of tests",
|
|
type=int, dest="max_tests", default=None, metavar="COUNT")
|
|
group.add_option("", "--pick-input",
|
|
help="Randomly select an input byte as well as fuzzing",
|
|
action='store_true', dest="pick_input", default=False)
|
|
parser.add_option_group(group)
|
|
|
|
parser.disable_interspersed_args()
|
|
|
|
(opts, args) = parser.parse_args()
|
|
|
|
if not args:
|
|
parser.error("Invalid number of arguments")
|
|
|
|
# Collect the list of inputs.
|
|
input_files = list(opts.input_files)
|
|
for filelist in opts.filelists:
|
|
f = open(filelist)
|
|
try:
|
|
for ln in f:
|
|
ln = ln.strip()
|
|
if ln:
|
|
input_files.append(ln)
|
|
finally:
|
|
f.close()
|
|
input_files.sort()
|
|
|
|
if not input_files:
|
|
parser.error("No input files!")
|
|
|
|
print '%s: note: fuzzing %d files.' % (sys.argv[0], len(input_files))
|
|
|
|
# Make sure the log directory exists if used.
|
|
if opts.log_dir:
|
|
if not os.path.exists(opts.log_dir):
|
|
try:
|
|
os.mkdir(opts.log_dir)
|
|
except OSError:
|
|
print "%s: error: log directory couldn't be created!" % (
|
|
sys.argv[0],)
|
|
raise SystemExit,1
|
|
|
|
# Get the list if insert/replacement strings.
|
|
replacements = list(opts.replacement_chars)
|
|
replacements.extend(opts.replacement_strings)
|
|
for replacement_list in opts.replacement_lists:
|
|
f = open(replacement_list)
|
|
try:
|
|
for ln in f:
|
|
ln = ln[:-1]
|
|
if ln:
|
|
replacements.append(ln)
|
|
finally:
|
|
f.close()
|
|
|
|
# Unique and order the replacement list.
|
|
replacements = list(set(replacements))
|
|
replacements.sort()
|
|
|
|
# Create the test generator.
|
|
tg = TestGenerator(input_files, opts.enable_delete, opts.enable_insert,
|
|
opts.enable_replace, replacements, opts.pick_input)
|
|
|
|
print '%s: note: %d input bytes.' % (sys.argv[0], tg.num_positions)
|
|
print '%s: note: %d total tests.' % (sys.argv[0], tg.num_tests)
|
|
if opts.test is not None:
|
|
it = [opts.test]
|
|
elif opts.max_tests is not None:
|
|
it = itertools.imap(random.randrange,
|
|
itertools.repeat(tg.num_tests, opts.max_tests))
|
|
else:
|
|
it = itertools.imap(random.randrange, itertools.repeat(tg.num_tests))
|
|
for test in it:
|
|
t = tg.get_test(test)
|
|
|
|
if opts.verbose:
|
|
print '%s: note: running test %d: %r' % (sys.argv[0], test, t)
|
|
ta = TestApplication(tg, t)
|
|
try:
|
|
ta.apply()
|
|
test_result = run_one_test(ta, test, input_files, args)
|
|
if not test_result and opts.stop_on_fail:
|
|
opts.revert = False
|
|
sys.exit(1)
|
|
finally:
|
|
if opts.revert:
|
|
ta.revert()
|
|
|
|
sys.stdout.flush()
|
|
|
|
if __name__ == '__main__':
|
|
main()
|