Add help message describing the mutators available and filtering.

Mutators can now be listed as part of a help command, and may then
be filtered by the user supplying a filter specification to disable
or enable only certain mutators.

pythonfuzz/corpus.py

@@ -351,7 +351,7 @@ class CorpusError(Exception):
 
 class Corpus(object):
 
-    def __init__(self, dirs=None, max_input_size=4096):
+    def __init__(self, dirs=None, max_input_size=4096, mutators_filter=None):
         self._inputs = []
         self._max_input_size = max_input_size
         self._dirs = dirs if dirs else []
@@ -370,15 +370,36 @@ def __init__(self, dirs=None, max_input_size=4096):
         self._seed_idx = 0
         self._save_corpus = dirs and os.path.isdir(dirs[0])
 
+        # Work out what we'll filter
+        filters = mutators_filter.split(' ')
+        negative_filters = [f[1:] for f in filters if f and f[0] == '!']
+        required_filters = [f for f in filters if f and f[0] != '!']
+
+        def acceptable(cls):
+            # No filters => everything's fine!
+            if mutators_filter is None:
+                return True
+
+            # First check that the required mutator types are set
+            for f in required_filters:
+                if f not in cls.types:
+                    return False
+            # Now remove any that are not allowed
+            for f in negative_filters:
+                if f in cls.types:
+                    return False
+
+            return True
+
         # Construct an object for each mutator we can use
-        self._mutators = [cls(self) for cls in mutator_classes]
-        if not self._mutators:
+        self.mutators = [cls(self) for cls in mutator_classes if acceptable(cls)]
+        if not self.mutators:
             raise CorpusError("No mutators are available")
 
     def __repr__(self):
         return "<{}(corpus of {}, %i mutators)>".format(self.__class__.__name__,
                                                         len(self._inputs),
-                                                        len(self._mutators))
+                                                        len(self.mutators))
 
     def _add_file(self, path):
         with open(path, 'rb') as f:
@@ -439,8 +460,8 @@ def mutate(self, buf):
 
             # Select a mutator from those we can apply
             while True:
-                x = self._rand(len(self._mutators))
-                mutator = self._mutators[x]
+                x = self._rand(len(self.mutators))
+                mutator = self.mutators[x]
 
                 newres = mutator.mutate(res)
                 if newres is not None:

pythonfuzz/fuzzer.py

@@ -88,22 +88,32 @@ def __init__(self,
                  regression=False,
                  max_input_size=4096,
                  close_fd_mask=0,
-                 runs=-1):
+                 runs=-1,
+                 mutators_filter=None):
         self._target = target
         self._dirs = [] if dirs is None else dirs
         self._exact_artifact_path = exact_artifact_path
         self._rss_limit_mb = rss_limit_mb
         self._timeout = timeout
         self._regression = regression
         self._close_fd_mask = close_fd_mask
-        self._corpus = corpus.Corpus(self._dirs, max_input_size)
+        self._corpus = corpus.Corpus(self._dirs, max_input_size, mutators_filter)
         self._total_executions = 0
         self._executions_in_sample = 0
         self._last_sample_time = time.time()
         self._total_coverage = 0
         self._p = None
         self.runs = runs
 
+    def help_mutators(self):
+        print("Mutators currently available (and their types):")
+        active_mutators = [mutator.__class__ for mutator in self._corpus.mutators]
+        for mutator in corpus.mutator_classes:
+            active = mutator in active_mutators
+            indicator = '-' if not active else ' '
+            print("  {}{:<60s} [{}]".format(indicator, mutator.name, ', '.join(sorted(mutator.types))))
+        print("\nMutators prefixed by '-' are currently disabled.")
+
     def log_stats(self, log_type):
         rss = (psutil.Process(self._p.pid).memory_info().rss + psutil.Process(os.getpid()).memory_info().rss) / 1024 / 1024
 

pythonfuzz/main.py

@@ -19,13 +19,19 @@ def __call__(self, *args, **kwargs):
         parser.add_argument('--max-input-size', type=int, default=4096, help='Max input size in bytes')
         parser.add_argument('--close-fd-mask', type=int, default=0, help='Indicate output streams to close at startup')
         parser.add_argument('--runs', type=int, default=-1, help='Number of individual test runs, -1 (the default) to run indefinitely.')
+        parser.add_argument('--help-mutators', action='store_true', help='Display help on the mutators')
+        parser.add_argument('--mutator-filter', type=str, default=None, help='Filter for mutator types to use; prefix with ! to disable')
         parser.add_argument('--timeout', type=int, default=30,
                             help='If input takes longer then this timeout the process is treated as failure case')
         args = parser.parse_args()
         f = fuzzer.Fuzzer(self.function, args.dirs, args.exact_artifact_path,
                           args.rss_limit_mb, args.timeout, args.regression, args.max_input_size,
-                          args.close_fd_mask, args.runs)
-        f.start()
+                          args.close_fd_mask, args.runs, args.mutator_filter)
+
+        if args.help_mutators:
+            f.help_mutators()
+        else:
+            f.start()
 
 
 if __name__ == '__main__':