Updating utils scripts

Author: software-opal

fuzz_to_test_cases.py

@@ -0,0 +1,62 @@
+import hashlib
+import multiprocessing
+import os
+import pathlib
+import subprocess
+import sys
+import functools
+import tempfile
+
+# Workflow taken from this blog post on how to fuzz effectively
+# https://foxglovesecurity.com/2016/03/15/fuzzing-workflows-a-fuzz-job-from-start-to-finish/
+
+
+ROOT = pathlib.Path(__file__).parent
+
+
+def list_files_in_folder(*folders):
+    folders = [folder for folder in folders if folder.is_dir()]
+    return {f for folder in folders for f in folder.iterdir() if f.is_file()}
+
+
+def load_files_in_folder(*folders):
+    return {f.read_bytes() for f in list_files_in_folder(*folders)}
+
+
+def minimise_inputs(executable, out_folder, items):
+    with tempfile.TemporaryDirectory(dir=ROOT) as _td:
+        td = pathlib.Path(_td)
+        assert td.is_dir()
+        for i, item in enumerate(sorted(items, key=lambda d: (len(d), d))):
+            if 0 < len(item) <= 4096:
+                (td / f"input_{i}").write_bytes(item)
+        subprocess.run(
+            ["afl-cmin", "-i", td, "-o", out_folder, "--", executable], check=True
+        )
+
+
+def minimise_input(executable, out_folder, file):
+    out = out_folder / file.name
+    subprocess.run(["afl-tmin", "-i", file, "-o", out, "--", executable], check=True)
+
+
+def main():
+    executable = pathlib.Path(sys.argv[1])
+    assert executable.is_file()
+    folders = [f for f in map(pathlib.Path, sys.argv[2:]) if f.is_dir()]
+    new_corpus = load_files_in_folder(*folders)
+    count = 0
+    cmin_out = ROOT / f"_cmin_{count}_corpus"
+    tmin_out = ROOT / f"_tmin_{count}_corpus"
+    subprocess.run(["rm", "-rf", cmin_out, tmin_out], check=True)
+    tmin_out.mkdir(exist_ok=True)
+    minimise_inputs(executable, cmin_out, new_corpus)
+    with multiprocessing.Pool(32) as p:
+        p.map(
+            functools.partial(minimise_input, executable, tmin_out),
+            list_files_in_folder(cmin_out),
+        )
+
+
+if __name__ == "__main__":
+    main()

queue_to_new_inputs.py

@@ -27,7 +27,7 @@
     101: "Rust Panic",
     134: "C Abort",
     -6: "C Abort",
-    999: "Python Timeout",
+    # 999: "Python Timeout",
 }
 
 KCOV_COMMON_OPTS = []
@@ -53,12 +53,13 @@ def kcov_collect(dir):
 TEST_OUTPUT_DIRS = {
     name: TEST_OUTPUT_ROOT_DIR / name
     for name in [
-        "match",
-        "match_err",
-        "match_crash",
+        'timeout',
         "crash",
-        "difference_out",
         "difference_err",
+        "difference_out",
+        "match_crash",
+        "match_err",
+        "match",
         "unknown",
         "wip",
     ]
@@ -201,10 +202,10 @@ def bytes_to_rust_array(data):
     for b in data:
         aparts.append(f"0x{b:02X},")
         if len(aparts) >= 16:
-            out += f'{"  ".join(aparts)}\n'
+            out += f'{" ".join(aparts)}\n'
             aparts = []
     if aparts:
-        out += f'{"  ".join(aparts)}\n'
+        out += f'{" ".join(aparts)}\n'
     return f"[\n{out}]\n"
 def output_test_case(
     input,
@@ -218,17 +219,20 @@ def output_test_case(
     fail_type,
 ):
     name = test_case_name(input)
+    if fail_type not in TEST_OUTPUT_DIRS:
+        print(f"You forgot a fail_type({fail_type!r}). I got you fam")
+        TEST_OUTPUT_DIRS[fail_type] = TEST_OUTPUT_ROOT_DIR / fail_type
     out = TEST_OUTPUT_DIRS[fail_type] / name
     sp_run(["rm", "-rf", out], check=True)
     out.mkdir(parents=True)
     (out / "input.dat").write_bytes(input)
     (out / "input.txt").write_text(bytes_to_test_hex(input))
-    (out / "input.txt").write_text(bytes_to_rust_array(input))
+    (out / "input.rs").write_text(bytes_to_rust_array(input))
     for (o, name) in [(sys_out, "sys_"), (new_out, "new_")]:
         if o is not None:
             (out / f"{name}output.dat").write_bytes(o)
             (out / f"{name}output.txt").write_text(bytes_to_test_hex(o))
-            (out / f"{name}output.txt").write_text(bytes_to_rust_array(o))
+            (out / f"{name}output.rs").write_text(bytes_to_rust_array(o))
     for (o, name) in [(sys_err, "sys_"), (new_err, "new_")]:
         if o is not None:
             (out / f"{name}err.txt").write_bytes(o)
@@ -281,7 +285,9 @@ def run(input: bytes):
     if not new_known_err:
         print("Unknown error: ", new_err)
 
-    if sys_rc == 0:
+    if new_rc == 999:
+        fail_type = 'timeout'
+    elif sys_rc == 0:
         if new_rc != 0:
             fail_type = "difference_err"
         elif sys_out != new_out:

test_case_gen.py

@@ -5,7 +5,7 @@
 import sys
 import tempfile
 
-ROOT = pathlib.Path(__file__).parent
+ROOT = pathlib.Path(__file__).parent.parent
 FUZZ_DIR = ROOT / "fuzz"
 FUZZ_CRASHES_DIR = FUZZ_DIR / "artifacts"
 AFL_OUTPUT_DIR = FUZZ_DIR / "afl_out"

file_to_test_case.py renamed to utils/file_to_test_case.py

@@ -5,7 +5,7 @@
 
 offset = [20, 24, 28]
 
-ROOT = pathlib.Path(__file__).parent
+ROOT = pathlib.Path(__file__).parent.parent
 NON_ALPHA = re.compile("(^[0-9]|[^a-zA-Z0-9_])")
 
 
@@ -47,7 +47,7 @@ def get_hus_vip_parts(data):
 
 
 def main():
-    out_folder = ROOT / "old_tests/"
+    out_folder = ROOT / "tests/"
     out_folder.mkdir(exist_ok=True)
     fuzz_inputs = ROOT / "_known_inputs"
     fuzz_inputs.mkdir(exist_ok=True)
@@ -57,6 +57,7 @@ def main():
         data = get_hus_vip_parts(p.read_bytes())
 
         with (out_folder / f"{name}.rs").open("w") as f:
+            f.write("#[macro_use]\nmod macros;\n\n")
             f.write("test_match_sys_decompress! {\n")
             for (name, compressed_data) in sorted(data.items()):
                 f.write(