| #!/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() |