Implement reproduce subcommand

Author: Merricx

src/zkregex_fuzzer/cli.py

@@ -7,16 +7,17 @@
 import random
 import uuid
 from pathlib import Path
+from zkregex_fuzzer.reproduce import reproduce
 from zkregex_fuzzer.fuzzer import fuzz_with_database, fuzz_with_grammar
 from zkregex_fuzzer.grammar import REGEX_GRAMMAR
 from zkregex_fuzzer.configs import TARGETS, VALID_INPUT_GENERATORS, GENERATORS
 from zkregex_fuzzer.harness import HarnessStatus
 from zkregex_fuzzer.logger import logger
 from zkregex_fuzzer.runner.circom import CircomSubprocess, SnarkjsSubprocess, ZkRegexSubprocess
 
-def main():
+def fuzz_parser():
     parser = argparse.ArgumentParser(
-        description="Generate fuzzed regexes using The Fuzzing Book's GrammarFuzzer."
+        add_help=False
     )
     parser.add_argument(
         "--regex-num",
@@ -93,17 +94,26 @@ def main():
         type=str,
         help="Path to the ptau (powers-of-tau) file for the proving step"
     )
-    parser.add_argument(
-        "--logger-level",
-        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
-        default="INFO",
-        help="Set the logger level (default: INFO)."
+    
+
+    return parser
+
+def reproduce_parser():
+    parser = argparse.ArgumentParser(
+        add_help=False
     )
 
-    args = parser.parse_args()
+    parser.add_argument(
+        "--path",
+        nargs="+",
+        type=str,
+        help="Path to the target directory output that want to be reproduced (support wildcard pattern).",
+        required=True
+    )
 
-    logger.setLevel(args.logger_level)
+    return parser
 
+def do_fuzz(args):
     if args.oracle == "valid" and not args.valid_input_generator:
         print("Valid input generator is required for valid oracle.")
         exit(1)
@@ -182,6 +192,41 @@ def main():
             inputs_num=args.inputs_num,
             kwargs=kwargs
         )
+
+def do_reproduce(args):
+    reproduce(args.path)
+
+def main():
+    
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--logger-level",
+        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
+        default="INFO",
+        help="Set the logger level (default: INFO)."
+    )
+
+    subparser = parser.add_subparsers(dest="subcommand")
+    subparser.add_parser(
+        "fuzz",
+        help="Generate fuzzed regexes using The Fuzzing Book's GrammarFuzzer.", 
+        parents=[fuzz_parser()]
+    )
+    subparser.add_parser(
+        "reproduce",
+        help="Reproduce the bug that found by the fuzzer.",
+        parents=[reproduce_parser()]
+    )
+
+    args = parser.parse_args()
+
+
+    logger.setLevel(args.logger_level)
+
+    if args.subcommand == "fuzz":
+        do_fuzz(args)
+    elif args.subcommand == "reproduce":
+        do_reproduce(args)
 
 
 if __name__ == "__main__":

src/zkregex_fuzzer/harness.py

@@ -7,12 +7,14 @@
 """
 
 import json
+import random
 from pathlib import Path
 from typing import Type, List, Union
 from enum import Enum
 from dataclasses import dataclass
 from zkregex_fuzzer.logger import logger
 from zkregex_fuzzer.runner import Runner, RegexCompileError, RegexRunError
+from zkregex_fuzzer.utils import get_random_filename
 
 class HarnessStatus(Enum):
     SUCCESS = 0 # Did not find a bug
@@ -40,26 +42,27 @@ def _return_harness_result(
         runner: Union[Runner, None],
         kwargs: dict,
     ):
-    if runner:
-        if result.status.name in status_to_save:
+    if result.status.name in status_to_save:
 
             metadata = {
-                "seed": kwargs.get("seed"),
-                "target": kwargs.get("target"),
-                "oracle": kwargs.get("oracle"),
-                "input_generator": kwargs.get("valid_input_generator"),
-                "fuzzer": kwargs.get("fuzzer"),
-                "regex_num": kwargs.get("regex_num"),
-                "inputs_num": kwargs.get("inputs_num"),
+                "config": kwargs,
+                "regex": result.regex,
+                "inputs": result.failed_inputs,
+                "status": result.status.name,
             }
 
-            dir_path = runner.save(output_path)
+            if runner:
+                dir_path = runner.save(output_path)
+            else:
+                dir_path = Path(output_path) / f"output_{get_random_filename()}"
+                dir_path.mkdir()
 
             metadata_json = json.dumps(metadata)
             metadata_path = Path(dir_path) / "metadata.json"
             with open(metadata_path.absolute(), "w") as f:
                 f.write(metadata_json)
 
+    if runner:
         runner.clean()
 
     return result
@@ -162,7 +165,7 @@ def harness(
         )
 
     return _return_harness_result(
-        HarnessResult(regex, inp_num, oracle, [], HarnessStatus.SUCCESS),
+        HarnessResult(regex, inp_num, oracle, inputs, HarnessStatus.SUCCESS),
         status_to_save,
         output_path,
         secondary_runner,

src/zkregex_fuzzer/reproduce.py

@@ -0,0 +1,95 @@
+"""
+Reproduce bugs found by the fuzzer.
+"""
+
+import glob
+import json
+from pathlib import Path
+from zkregex_fuzzer.configs import TARGETS
+from zkregex_fuzzer.logger import logger
+from zkregex_fuzzer.harness import HarnessStatus
+from zkregex_fuzzer.runner import RegexCompileError, RegexRunError
+from zkregex_fuzzer.utils import pretty_regex
+
+def reproduce(path_list: list[str]):
+
+    for pattern in path_list:
+        expanded_path = glob.glob(pattern)
+        if not expanded_path:
+            logger.info(f"Path {pattern} is not exist, skipping.")
+            continue
+
+        for path in expanded_path:
+            directory = Path(path)
+            if not directory.exists():
+                continue
+
+            if not (directory / "metadata.json").exists():
+                logger.info(f"metadata.json is not exist in {directory}, skipping.")
+                continue
+
+            simulate_harness(directory)
+            
+                    
+def simulate_harness(directory: Path):
+
+    with open(str(directory / "metadata.json"), 'r') as f:
+        metadata = json.loads(f.read())
+
+    regex = metadata["regex"]
+    inputs = metadata["inputs"]
+    expected_status = metadata["status"]
+    kwargs = metadata["config"]
+
+    target_runner = TARGETS[metadata["config"]["target"]]
+    oracle = True if metadata["config"]["oracle"] == "valid" else False
+
+    print(f"Reproducing regex: {pretty_regex(regex)}")
+    print("-" * 80)
+    print(f"Directory path: {directory}")
+    print(f"Inputs: {inputs}")
+    print(f"Expected result: {expected_status}")
+    print("-" * 80)
+
+    try:
+        runner = target_runner(regex, kwargs)
+    except RegexCompileError as e:
+        if expected_status == HarnessStatus.COMPILE_ERROR.name:
+            print("Reproduce completed successfully!")
+            print("-" * 80)
+            print(e)
+        else:
+            print(f"Unexpected COMPILE_ERROR reproduce status: {e}")
+
+        print("=" * 80)
+        return
+
+    failed_inputs = []
+    for input in inputs:
+
+        try:
+            runner_status, runner_str = runner.match(input)
+            if runner_status != oracle:
+                failed_inputs.append(input)
+
+        except RegexRunError as e:
+            if expected_status == HarnessStatus.RUN_ERROR.name:
+                print("Reproduce completed successfully!")
+                print("-" * 80)
+                print(e)
+            else:
+                print(f"Unexpected RUN_ERROR reproduce status: {e}")
+
+    if len(failed_inputs) > 0:
+        if expected_status == HarnessStatus.FAILED.name:
+            print("Reproduce completed successfully!")
+        else:
+            print(f"Unexpected FAILED reproduce status: {failed_inputs}")
+    else:
+        if expected_status == HarnessStatus.SUCCESS.name:
+            print("Reproduce completed successfully!")
+        else:
+            print(f"Unexpected SUCCESS reproduce status: {inputs}")
+
+    runner.clean()
+    print("=" * 80)

src/zkregex_fuzzer/runner/circom.py

@@ -355,10 +355,10 @@ def save(self, path) -> str:
         target_path = Path(path).resolve() / f"output_{r1cs_path.stem}"
         target_path.mkdir()
 
-        circom_path.replace(target_path / circom_path.name)
-        input_path.replace(target_path / input_path.name)
-        r1cs_path.replace(target_path / r1cs_path.name)
-        wasm_path.replace(target_path / wasm_path.name)
+        if self._circom_path: circom_path.replace(target_path / circom_path.name)
+        if self._input_path: input_path.replace(target_path / input_path.name)
+        if self._r1cs_path: r1cs_path.replace(target_path / r1cs_path.name)
+        if self._wasm_path: wasm_path.replace(target_path / wasm_path.name)
 
         if self._run_the_prover:
             zkey_path = Path(self._zkey_path)

src/zkregex_fuzzer/utils.py

@@ -3,7 +3,8 @@
 """
 
 import re
-
+import random
+import string
 from fuzzingbook.Grammars import simple_grammar_fuzzer, Grammar
 
 
@@ -98,6 +99,9 @@ def grammar_fuzzer(grammar: Grammar, start_symbol: str, max_nonterminals: int =
                            max_nonterminals=max_nonterminals,
                            max_expansion_trials=max_expansion_trials)
 
+def get_random_filename():
+    return ''.join(random.choices(string.ascii_lowercase + string.digits, k=8))
+
 def pretty_regex(regex: str):
     """
     Format raw string regex to printable chars