Merge pull request #29 from Rog3rSm1th/dev/rog3rsm1th

Dev/rog3rsm1th

README.md

@@ -2,7 +2,7 @@
   <img src="https://github.com/Rog3rSm1th/Frelatage/blob/main/doc/frelatage_logo.gif?raw=true" width="200" height="200" style="border-radius:4px"/>
   <br>
   <code>pip3 install frelatage</code></br>
-  <i>Current release : <a href="https://github.com/Rog3rSm1th/Frelatage/releases">0.0.8</a></i></br></br>
+  <i>Current release : <a href="https://github.com/Rog3rSm1th/Frelatage/releases">0.1.0</a></i></br></br>
   <a target="_blank" href="https://www.python.org/downloads/" title="Python version"><img src="https://img.shields.io/badge/Made%20with-Python-1f425f.svg"></a>
   <a target="_blank" href="https://www.python.org/downloads/" title="Python version"><img src="https://img.shields.io/badge/python-%3E=_3.6-green.svg"></a>
   <a target="_blank" href="LICENSE" title="License: MIT"><img src="https://img.shields.io/badge/License-MIT-blue.svg"></a>
@@ -263,7 +263,9 @@ f = frelatage.Fuzzer(
     # Directory where the error reports will be stored
     output_directory="./out",
     # Enable or disable silent mode
-    silent=False
+    silent=False,
+    # Enable or disable infinite fuzzing
+    infinite_fuzz=False
 )
 f.fuzz()
 ```

examples/yaml_fuzzer/yaml_fuzzer_dump.py

@@ -16,6 +16,6 @@ def fuzz_yaml_dump(data):
 })
 
 # Initialize the fuzzer
-f = frelatage.Fuzzer(fuzz_yaml_dump, [[dictionary]])
+f = frelatage.Fuzzer(fuzz_yaml_dump, [[dictionary]], infinite_fuzz=True)
 # Fuzz
 f.fuzz()

frelatage/__init__.py

@@ -5,9 +5,10 @@
 from datetime import datetime
 from importlib.metadata import version
 from pathlib import Path
-from typing import Callable, List
+from typing import Callable
 from frelatage.corpus.corpus import load_corpus
 from frelatage.config.config import Config
+from frelatage.input.input import Input
 from frelatage.mutator.mutator import *
 from frelatage.queue.queue import Queue
 from frelatage.tracer.tracer import Tracer
@@ -30,28 +31,29 @@ class Fuzzer(object):
     of these fuzzers and gather them together into a new tool in order to efficiently fuzz python applications.
     """
 
-    from ._mutation import valid_mutators, get_mutation, generate_cycle_mutations
-    from ._interface import (
+    from ._mutation import valid_mutators, get_mutation, generate_cycle_mutations  # type: ignore
+    from ._interface import (  # type: ignore
         init_interface,
         refresh_interface,
         start_interface,
         exit_message,
     )
-    from ._evaluate import evaluate_mutations
-    from ._cycle import run_function, run_cycle
-    from ._fuzz import fuzz
-    from ._report import get_report_name, save_report
-    from ._input import init_input_folder, init_file_input_arguments, init_file_inputs
+    from ._evaluate import evaluate_mutations  # type: ignore
+    from ._cycle import run_function, run_cycle  # type: ignore
+    from ._fuzz import fuzz  # type: ignore
+    from ._report import get_report_name, save_report  # type: ignore
+    from ._input import init_input_folder, init_file_input_arguments, init_file_inputs  # type: ignore
 
     def __init__(
         self,
         method: Callable,
-        corpus: list[object],
+        corpus: list,
         threads_count: int = 8,
-        exceptions_whitelist: list = (),
-        exceptions_blacklist: list = (),
+        exceptions_whitelist: tuple = (),
+        exceptions_blacklist: tuple = (),
         output_directory: str = "./out",
         silent: bool = False,
+        infinite_fuzz: bool = False,
     ) -> None:
         """
         Initialize the fuzzer
@@ -62,13 +64,13 @@ def __init__(
         self.config = Config
 
         # Global set of reached instructions
-        self.reached_instructions = set([])
+        self.reached_instructions: set = set([])
         # Global set of instructions pairs
-        self.instructions_pairs = set([])
+        self.instructions_pairs: set = set([])
         # Global set of instruction pairs executed during a crash
-        self.favored_pairs = set([])
+        self.favored_pairs: set = set([])
         # Global Set of positions where a crash occurred
-        self.error_positions = set([])
+        self.error_positions: set = set([])
         # Fuzzed method
         self.method = method
 
@@ -79,9 +81,11 @@ def __init__(
         # List of all avalaibles mutators
         self.mutators = mutators
         # Number of concurrently launched threads
-        self.threads_count = max(min(threads_count, Config.FRELATAGE_MAX_THREADS), 8)
+        self.threads_count = max(
+            min(threads_count, Config.FRELATAGE_MAX_THREADS), 8
+        )
         # List of cycle mutations
-        self.cycle = []
+        self.cycle: list = []
 
         # Exceptions that will be taken into account or not when fuzzing
         self.exceptions_whitelist = exceptions_whitelist
@@ -96,17 +100,22 @@ def __init__(
         # Input and output directories
         # The working directory is the same as the fuzz file
         self.input_directory = os.path.join(
-            os.path.dirname(os.path.realpath(sys.argv[0])), Config.FRELATAGE_INPUT_DIR
+            os.path.dirname(os.path.realpath(sys.argv[0])),
+            Config.FRELATAGE_INPUT_DIR,
         )
         self.output_directory = Path(
             os.path.join(
-                os.path.dirname(os.path.realpath(sys.argv[0])), output_directory
+                os.path.dirname(os.path.realpath(sys.argv[0])),
+                output_directory,
             )
         ).as_posix()
 
         # Silent output
         self.silent = silent
 
+        # Infinite fuzzing
+        self.infinite_fuzz = infinite_fuzz
+
         # Fuzzer statistics
         self.cycles_count = 0
         self.inputs_count = 0

frelatage/_cycle.py

@@ -15,8 +15,12 @@ def run_function(self, arguments: list, result: list) -> bool:
     trace_instructions_count = len(trace_instructions)
     trace_instruction_pairs_count = len(trace_instruction_pairs)
 
-    new_instructions_count = len(trace_instructions - self.reached_instructions)
-    new_sequences_count = len(trace_instruction_pairs - self.instructions_pairs)
+    new_instructions_count = len(
+        trace_instructions - self.reached_instructions
+    )
+    new_sequences_count = len(
+        trace_instruction_pairs - self.instructions_pairs
+    )
 
     new_instruction_error = trace.error_position not in self.error_positions
 
@@ -40,7 +44,7 @@ def run_cycle(self) -> list[Report]:
     run the function with each mutation of the cycle as argument.
     Return the execution reports.
     """
-    cycle_reports = []
+    cycle_reports: list = []
 
     # TODO: Implement multithreading
     for mutation in self.cycle:

frelatage/_fuzz.py

@@ -6,6 +6,13 @@ def fuzz(self) -> None:
     """
     Run the fuzzer
     """
+    # Infinite fuzzing is allowed if we have one input combination
+    if self.infinite_fuzz and len(self.queue.arguments) > 1:
+        print(
+            "Error: infinite fuzzing is only possible with a corpus of size 1"
+        )
+        exit(1)
+
     try:
         # Interface
         if not self.silent:

frelatage/_input.py

@@ -77,7 +77,7 @@ def init_file_input_arguments(self) -> bool:
     return True
 
 
-def init_file_inputs(self) -> bool:
+def init_file_inputs(self) -> None:
     """
     Set up the tree structure to fuzz a function with "file" type arguments
     """

frelatage/_interface.py

@@ -1,7 +1,7 @@
 import curses
 import time
 from curses import wrapper
-from datetime import datetime
+from datetime import datetime, timedelta
 from string import Formatter
 from frelatage import __version__, Config
 from frelatage.colors import Colors
@@ -10,7 +10,7 @@
 REFRESH_INTERVAL = 0.1
 
 
-def format_delta(time_delta: datetime, format: str) -> str:
+def format_delta(time_delta: timedelta, format: str) -> str:
     """
     Format a time delta.
     """
@@ -79,10 +79,12 @@ def refresh_interface(self):
     # Process timing
     run_time = format_time_elapsed(self.fuzz_start_time).ljust(32)
     last_new_path_time = format_time_elapsed(self.last_new_path_time).ljust(32)
-    last_unique_crash_time = format_time_elapsed(self.last_unique_crash_time).ljust(32)
-    last_unique_timeout_time = format_time_elapsed(self.last_unique_timeout_time).ljust(
-        32
-    )
+    last_unique_crash_time = format_time_elapsed(
+        self.last_unique_crash_time
+    ).ljust(32)
+    last_unique_timeout_time = format_time_elapsed(
+        self.last_unique_timeout_time
+    ).ljust(32)
 
     # Overall results
     uniques_crashes_count = str(self.unique_crashes).ljust(9)
@@ -105,7 +107,8 @@ def refresh_interface(self):
         crashes=str(self.total_crashes), uniques=str(self.unique_crashes)
     ).ljust(22)
     total_timeouts = "{total_timeouts} [{timeout_delay} sec]".format(
-        total_timeouts=self.total_timeouts, timeout_delay=Config.FRELATAGE_TIMEOUT_DELAY
+        total_timeouts=self.total_timeouts,
+        timeout_delay=Config.FRELATAGE_TIMEOUT_DELAY,
     ).ljust(22)
 
     # Progress
@@ -116,7 +119,8 @@ def refresh_interface(self):
     current_argument = self.queue.position + 1
     total_arguments_count = len(self.queue.arguments)
     current_stage = "{current_argument}/{total_arguments_count}".format(
-        current_argument=current_argument, total_arguments_count=total_arguments_count
+        current_argument=current_argument,
+        total_arguments_count=total_arguments_count,
     ).ljust(16)
     stage_executions = str(self.stage_inputs_count).ljust(16)
 
@@ -136,7 +140,7 @@ def refresh_interface(self):
     │ Uniques crashes     :: {uniques_crashes_count}│ Cycles done         :: {cycles_count}│ Stage       :: {current_stage}│
     │ Unique timeouts     :: {uniques_timeouts_count}│ Total executions    :: {total_executions}│ Stage execs :: {stage_executions}│
     └─────────────────────────────────┴────────────────────────────────────┴────────────────────────────────┘
-    [ {execs_per_second} exec/s ]                                    
+    [ {execs_per_second} exec/s ]                             
     """.format(
             title=title,
             execs_per_second=execs_per_second,
@@ -168,10 +172,9 @@ def exit_message(
     run_time = format_time_elapsed(self.fuzz_start_time)
     uniques_crashes_count = str(self.unique_crashes)
     uniques_timeouts_count = str(self.unique_timeout)
-    total_crashes = "{crashes} ({uniques} uniques)".format(
-        crashes=str(self.total_crashes), uniques=str(self.unique_crashes)
+    total_timeouts = "{total_timeouts}".format(
+        total_timeouts=self.total_timeouts
     )
-    total_timeouts = "{total_timeouts}".format(total_timeouts=self.total_timeouts)
     total_paths_count = str(len(self.reached_instructions))
     cycles_count = str(self.cycles_count)
     total_executions = str(self.inputs_count)
@@ -240,7 +243,7 @@ def exit_message(
     return True
 
 
-def start_interface(self):
+def start_interface(self) -> None:
     """
     Display the curse interface
     """

frelatage/_mutation.py

@@ -1,7 +1,7 @@
 import copy
 import os
-import sys
 import random
+import sys
 from pathlib import Path
 from typing import Type, Any
 from frelatage.mutator.mutator import Mutator
@@ -30,7 +30,8 @@ def valid_mutators(self, input_type: Type, input_size: int) -> list[Mutator]:
     # Filtering mutators using the "allowed types" property of each mutator
     valid_mutators_type = list(
         filter(
-            lambda mutator: type_str in mutator.allowed_types and mutator.enabled,
+            lambda mutator: type_str in mutator.allowed_types
+            and mutator.enabled,
             self.mutators,
         )
     )
@@ -40,7 +41,10 @@ def valid_mutators(self, input_type: Type, input_size: int) -> list[Mutator]:
     if max_input_size_reached:
         # Filtering mutators using the "size_effect" property of each mutator
         valid_mutators_size = list(
-            filter(lambda mutator: "decrease" in mutator.size_effect, self.mutators)
+            filter(
+                lambda mutator: "decrease" in mutator.size_effect,
+                self.mutators,
+            )
         )
     else:
         valid_mutators_size = self.mutators
@@ -105,7 +109,7 @@ def get_mutation(self, input: Any, file: bool) -> Any:
     return mutation
 
 
-def generate_cycle_mutations(self, parents: list) -> list:
+def generate_cycle_mutations(self, parents: list) -> None:
     """
     Generate a list of mutations with a list of parent mutations as an input
     """
@@ -127,7 +131,9 @@ def generate_cycle_mutations(self, parents: list) -> list:
             # Mutation of "file" type inputs
             if mutation.file:
                 filename = os.path.split(mutation.value)[1]
-                file_argument_id = os.path.split(os.path.split(mutation.value)[0])[1]
+                file_argument_id = os.path.split(
+                    os.path.split(mutation.value)[0]
+                )[1]
                 base = Path(mutation.value).parents[2]
                 new_argument = os.path.join(
                     base, str(thread), file_argument_id, filename

frelatage/_report.py

@@ -20,9 +20,9 @@ def get_report_name(self, report: Report) -> str:
         else str(None)
     )
     # File where the error occured
-    error_file = os.path.splitext(os.path.basename(report.trace.error_position[0][0]))[
-        0
-    ].lower()
+    error_file = os.path.splitext(
+        os.path.basename(report.trace.error_position[0][0])
+    )[0].lower()
 
     report_name = "id:{error_id},err:{error_type},err_file:{error_file},err_pos:{error_position}".format(
         error_id=error_id,
@@ -65,7 +65,8 @@ def save_report(self, report) -> bool:
 
     # Save report file
     with open(
-        "{report_directory}/input".format(report_directory=report_directory), "wb+"
+        "{report_directory}/input".format(report_directory=report_directory),
+        "wb+",
     ) as f:
         # We use pickle to store the report object
         pickle.dump(custom_report, f)
@@ -79,7 +80,8 @@ def save_report(self, report) -> bool:
             file_argument_content = open(argument.value, "rb").read()
             # Save file in /out/<report name>/<argument number>/<file name>
             argument_directory = "{report_directory}/{argument_number}".format(
-                report_directory=report_directory, argument_number=argument_number
+                report_directory=report_directory,
+                argument_number=argument_number,
             )
             # create /out/<report name>/<argument number> folder if not exists
             if not os.path.exists(argument_directory):

frelatage/config/config.py

@@ -14,7 +14,9 @@ class Config:
 
     # Enable the use of mutations based on dictionary elements
     if os.getenv("FRELATAGE_DICTIONARY_ENABLE", "1") not in ("1", "0"):
-        raise FrelatageConfigError("FRELATAGE_DICTIONARY_ENABLE must '1' or '0'")
+        raise FrelatageConfigError(
+            "FRELATAGE_DICTIONARY_ENABLE must '1' or '0'"
+        )
     FRELATAGE_DICTIONARY_ENABLE = (
         True if os.getenv("FRELATAGE_DICTIONARY_ENABLE", "1") == "1" else False
     )
@@ -29,7 +31,9 @@ class Config:
     # Delay in seconds after which a function will return a timeoutError
     # Must be in range 1~20
     if not 1 <= int(os.getenv("FRELATAGE_TIMEOUT_DELAY", 2)) <= math.inf:
-        raise FrelatageConfigError("FRELATAGE_TIMEOUT_DELAY must be in range 1~20")
+        raise FrelatageConfigError(
+            "FRELATAGE_TIMEOUT_DELAY must be in range 1~20"
+        )
     FRELATAGE_TIMEOUT_DELAY = int(os.getenv("FRELATAGE_TIMEOUT_DELAY", 2))
 
     # Temporary folder where input files are stored.
@@ -41,13 +45,17 @@ class Config:
     # Maximum size of an input variable in bytes
     # Must be in range 4~1000000
     if not 4 <= int(os.getenv("FRELATAGE_INPUT_MAX_LEN", 4096)) <= math.inf:
-        raise FrelatageConfigError("FRELATAGE_INPUT_MAX_LEN must be in range 4~1000000")
+        raise FrelatageConfigError(
+            "FRELATAGE_INPUT_MAX_LEN must be in range 4~1000000"
+        )
     FRELATAGE_INPUT_MAX_LEN = int(os.getenv("FRELATAGE_INPUT_MAX_LEN", 4096))
 
     # Maximum number of simultaneous threads
     # Must be in range 8~50
     if not 8 <= int(os.getenv("FRELATAGE_MAX_THREADS", 20)) <= math.inf:
-        raise FrelatageConfigError("FRELATAGE_MAX_THREADS must be in range 8~50")
+        raise FrelatageConfigError(
+            "FRELATAGE_MAX_THREADS must be in range 8~50"
+        )
     FRELATAGE_MAX_THREADS = int(os.getenv("FRELATAGE_MAX_THREADS", 20))
 
     # Maximum number of successives cycle without a new path