Implement fuzzing driver using cobrafuzz

Ref. eng/recordflux/RecordFlux#1305

.gitignore

@@ -22,3 +22,4 @@ htmlcov
 pyproject.toml
 why3session.xml.bak
 why3shapes.gz.bak
+/crashes

Makefile

@@ -149,7 +149,7 @@ test_property:
 	$(PYTEST) tests/property
 
 test_tools:
-	$(PYTEST) tests/tools
+	$(PYTEST) --cov=tools --cov-branch --cov-fail-under=43.6 --cov-report=term-missing:skip-covered tests/tools
 
 test_ide:
 	$(PYTEST) tests/ide
@@ -194,6 +194,10 @@ test_installation: $(SDIST)
 	HOME=$(BUILD_DIR)/test_installation $(BUILD_DIR)/venv/bin/rflx install gnatstudio
 	test -f $(BUILD_DIR)/test_installation/.gnatstudio/plug-ins/recordflux.py
 
+fuzz_parser: FUZZER_RUNS=-1
+fuzz_parser:
+	./tools/fuzz_driver.py --state-dir $(BUILD_DIR)/fuzzer --corpus-dir $(MAKEFILE_DIR) --runs=$(FUZZER_RUNS)
+
 .PHONY: prove prove_tests prove_python_tests prove_apps prove_property_tests
 
 prove: prove_tests prove_python_tests prove_apps
@@ -335,7 +339,7 @@ $(GENERATED_DIR)/python/librflxlang/librflxlang.so: $(wildcard language/*.py) |
 .PHONY: clean clean_all
 
 clean:
-	rm -rf $(BUILD_DIR)/[^_]* .coverage .coverage.* .hypothesis .mypy_cache .pytest_cache .ruff_cache
+	rm -rf $(BUILD_DIR)/[^_]* .coverage .coverage.* .hypothesis .mypy_cache .pytest_cache .ruff_cache crashes
 	$(MAKE) -C examples/apps/ping clean
 	$(MAKE) -C examples/apps/dhcp_client clean
 	$(MAKE) -C examples/apps/spdm_responder clean

setup.py

@@ -98,6 +98,7 @@ def run(self) -> None:
     extras_require={
         "devel": [
             "build >= 0.9, <1",
+            "cobrafuzz ==1.0.12",
             "furo == 2022.4.7",
             "hypothesis >=6.14, <6.24",
             "lark ==1.1.8",

tests/tools/test_fuzz_driver.py

@@ -0,0 +1,114 @@
+import logging
+import sys
+from pathlib import Path
+
+import pytest
+
+import rflx.specification.parser
+from tests.const import SPEC_DIR
+from tools import fuzz_driver
+
+
+class UnexpectedError(Exception):
+    pass
+
+
+def test_no_bug(
+    monkeypatch: pytest.MonkeyPatch,
+    tmpdir: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with monkeypatch.context() as mp:
+        mp.setattr(
+            sys,
+            "argv",
+            [
+                "fuzz_driver.py",
+                "--state-dir",
+                str(tmpdir),
+                "--corpus-dir",
+                str(SPEC_DIR),
+                "--runs",
+                "10",
+            ],
+        )
+
+        with caplog.at_level(logging.INFO), pytest.raises(SystemExit, match="^0$"):
+            fuzz_driver.main()
+        assert "did 10 runs, stopping now." in caplog.text
+
+
+def test_unexpected_error(
+    monkeypatch: pytest.MonkeyPatch,
+    tmpdir: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    def raise_unexpected_exception() -> None:
+        raise UnexpectedError("Unexpected error")
+
+    with monkeypatch.context() as mp:
+        mp.setattr(
+            sys,
+            "argv",
+            [
+                "fuzz_driver.py",
+                "--state-dir",
+                str(tmpdir),
+                "--corpus-dir",
+                str(SPEC_DIR),
+                "--artifact-file",
+                str(tmpdir / "crash"),
+                "--runs",
+                "10",
+            ],
+        )
+        mp.setattr(
+            rflx.specification.Parser,
+            "parse_string",
+            lambda _c, _s: raise_unexpected_exception(),
+        )
+
+        with caplog.at_level(logging.INFO), pytest.raises(SystemExit, match="^76$"):
+            fuzz_driver.main()
+        assert f'sample was written to {tmpdir/ "crash"}' in caplog.text
+
+
+def test_decode_error(
+    monkeypatch: pytest.MonkeyPatch,
+    tmpdir: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    def raise_decode_error() -> None:
+        raise UnicodeDecodeError(
+            "fakeenc",
+            b"deafbeef",
+            1,
+            2,
+            "Error",
+        )
+
+    with monkeypatch.context() as mp:
+        mp.setattr(
+            sys,
+            "argv",
+            [
+                "fuzz_driver.py",
+                "--state-dir",
+                str(tmpdir),
+                "--corpus-dir",
+                str(SPEC_DIR),
+                "--artifact-file",
+                str(tmpdir / "crash"),
+                "--runs",
+                "10",
+            ],
+        )
+        mp.setattr(
+            rflx.specification.Parser,
+            "parse_string",
+            lambda _c, _s: raise_decode_error(),
+        )
+
+        with caplog.at_level(logging.INFO), pytest.raises(SystemExit, match="^0$"):
+            fuzz_driver.main()
+        assert "did 10 runs, stopping now." in caplog.text

tools/fuzz_driver.py

@@ -0,0 +1,85 @@
+#!/usr/bin/env python3
+
+import argparse
+import random
+import sys
+from pathlib import Path
+
+from cobrafuzz.fuzzer import Fuzzer
+
+from rflx import error
+from rflx.model import Cache, Digest
+from rflx.specification import parser
+
+
+class SilentlyNeverVerify(Cache):
+    def __init__(self) -> None:
+        pass
+
+    def is_verified(self, _: Digest) -> bool:
+        return True
+
+    def add_verified(self, digest: Digest) -> None:
+        pass
+
+
+def fuzz(buf: bytes) -> None:
+    try:
+        string = buf.decode("utf-8")
+        p = parser.Parser(cache=SilentlyNeverVerify())
+        p.parse_string(string)
+        p.create_model()
+    except (UnicodeDecodeError, error.RecordFluxError):
+        pass
+    except KeyboardInterrupt:  # pragma: no cover
+        sys.exit()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--state-dir",
+        type=Path,
+        required=True,
+        help="Directory to hold fuzzer state",
+    )
+    parser.add_argument(
+        "--corpus-dir",
+        type=Path,
+        required=True,
+        help="Directory to extract corpus from",
+    )
+    parser.add_argument(
+        "--artifact-file",
+        type=Path,
+        help="Store single artifact in given file",
+    )
+    parser.add_argument(
+        "--runs",
+        type=int,
+        default=-1,
+        help="Maxium number of runs",
+    )
+    parser.add_argument(
+        "--timeout",
+        type=int,
+        default=30,
+        help="Time after which an execution is considered a hang",
+    )
+    arguments = parser.parse_args()
+
+    corpus = [Path(p) for p in arguments.corpus_dir.glob("**/*.rflx")]
+    random.shuffle(corpus)
+
+    fuzzer = Fuzzer(
+        fuzz,
+        dirs=[Path(arguments.state_dir), *corpus],
+        timeout=arguments.timeout,
+        runs=arguments.runs,
+        exact_artifact_path=arguments.artifact_file,
+    )
+    fuzzer.start()
+
+
+if __name__ == "__main__":
+    main()  # pragma: no cover