fuzz: initial integration (#117)

* fuzz: initial integration

* ci: skip atheris for 3.12

Author: manunio

fuzz/README.md

@@ -0,0 +1,37 @@
+# Fuzz Testing
+
+Fuzz testing is:
+
+> An automated software testing technique that involves providing invalid, unexpected, or random data as inputs to a program.
+
+We use coverage guided fuzz testing to automatically discover bugs in python-multipart.
+
+This `fuzz/` directory contains the configuration and the fuzz tests for python-multipart.
+To generate and run fuzz tests, we use the [Atheris](https://github.com/google/atheris) library.
+
+## Running a fuzzer
+
+This directory contains fuzzers like for example `fuzz_form.py`. You can run it with:
+
+Run fuzz target:
+```sh
+$ python fuzz/fuzz_form.py
+```
+
+You should see output that looks something like this:
+
+```
+#2      INITED cov: 32 ft: 32 corp: 1/1b exec/s: 0 rss: 49Mb
+#3      NEW    cov: 33 ft: 33 corp: 2/2b lim: 4 exec/s: 0 rss: 49Mb L: 1/1 MS: 1 ChangeByte-
+#4      NEW    cov: 97 ft: 97 corp: 3/4b lim: 4 exec/s: 0 rss: 49Mb L: 2/2 MS: 1 InsertByte-
+#11     NEW    cov: 116 ft: 119 corp: 4/5b lim: 4 exec/s: 0 rss: 49Mb L: 1/2 MS: 2 ChangeBinInt-EraseBytes-
+#30     NEW    cov: 131 ft: 134 corp: 5/8b lim: 4 exec/s: 0 rss: 49Mb L: 3/3 MS: 4 ChangeByte-ChangeBit-InsertByte-CopyPart-
+#31     NEW    cov: 135 ft: 138 corp: 6/11b lim: 4 exec/s: 0 rss: 49Mb L: 3/3 MS: 1 CrossOver-
+#39     NEW    cov: 135 ft: 142 corp: 7/15b lim: 4 exec/s: 0 rss: 49Mb L: 4/4 MS: 3 ChangeBit-CrossOver-CopyPart-
+```
+
+It will continue to generate random inputs forever, until it finds a
+bug or is terminated. The testcases for bugs it finds can be seen in
+the form of `crash-*` or `timeout-*` at the place from where command is run.
+You can rerun the fuzzer on a single input by passing it on the
+command line `python fuzz/fuzz_form.py /path/to/testcase`.

fuzz/corpus/fuzz_decoders/fuzz_decoders

@@ -0,0 +1,40 @@
+import io
+import sys
+
+import atheris
+from helpers import EnhancedDataProvider
+
+with atheris.instrument_imports():
+    from multipart.decoders import Base64Decoder, DecodeError, QuotedPrintableDecoder
+
+
+def fuzz_base64_decoder(fdp: EnhancedDataProvider) -> None:
+    decoder = Base64Decoder(io.BytesIO())
+    decoder.write(fdp.ConsumeRandomBytes())
+    decoder.finalize()
+
+
+def fuzz_quoted_decoder(fdp: EnhancedDataProvider) -> None:
+    decoder = QuotedPrintableDecoder(io.BytesIO())
+    decoder.write(fdp.ConsumeRandomBytes())
+    decoder.finalize()
+
+
+def TestOneInput(data: bytes) -> None:
+    fdp = EnhancedDataProvider(data)
+    targets = [fuzz_base64_decoder, fuzz_quoted_decoder]
+    target = fdp.PickValueInList(targets)
+
+    try:
+        target(fdp)
+    except DecodeError:
+        return
+
+
+def main():
+    atheris.Setup(sys.argv, TestOneInput)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()

fuzz/corpus/fuzz_form/fuzz_form

@@ -0,0 +1,53 @@
+import io
+import sys
+from unittest.mock import Mock
+
+import atheris
+from helpers import EnhancedDataProvider
+
+with atheris.instrument_imports():
+    from multipart.exceptions import FormParserError
+    from multipart.multipart import parse_form
+
+on_field = Mock()
+on_file = Mock()
+
+
+def parse_octet_stream(fdp: EnhancedDataProvider) -> None:
+    header = {"Content-Type": "application/octet-stream"}
+    parse_form(header, io.BytesIO(fdp.ConsumeRandomBytes()), on_field, on_file)
+
+
+def parse_url_encoded(fdp: EnhancedDataProvider) -> None:
+    header = {"Content-Type": "application/x-url-encoded"}
+    parse_form(header, io.BytesIO(fdp.ConsumeRandomBytes()), on_field, on_file)
+
+
+def parse_form_urlencoded(fdp: EnhancedDataProvider) -> None:
+    header = {"Content-Type": "application/x-www-form-urlencoded"}
+    parse_form(header, io.BytesIO(fdp.ConsumeRandomBytes()), on_field, on_file)
+
+
+def parse_multipart_form_data(fdp: EnhancedDataProvider) -> None:
+    header = {"Content-Type": "multipart/form-data; boundary=--boundary"}
+    parse_form(header, io.BytesIO(fdp.ConsumeRandomBytes()), on_field, on_file)
+
+
+def TestOneInput(data: bytes) -> None:
+    fdp = EnhancedDataProvider(data)
+    targets = [parse_octet_stream, parse_url_encoded, parse_form_urlencoded, parse_multipart_form_data]
+    target = fdp.PickValueInList(targets)
+
+    try:
+        target(fdp)
+    except FormParserError:
+        return
+
+
+def main():
+    atheris.Setup(sys.argv, TestOneInput)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()

fuzz/corpus/fuzz_options_header/fuzz_options_header

@@ -0,0 +1,26 @@
+import sys
+
+import atheris
+from helpers import EnhancedDataProvider
+
+with atheris.instrument_imports():
+    from multipart.multipart import parse_options_header
+
+
+def TestOneInput(data: bytes) -> None:
+    fdp = EnhancedDataProvider(data)
+    try:
+        parse_options_header(fdp.ConsumeRandomBytes())
+    except AssertionError:
+        return
+    except TypeError:
+        return
+
+
+def main():
+    atheris.Setup(sys.argv, TestOneInput)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()

fuzz/fuzz_decoders.py

@@ -0,0 +1,9 @@
+import atheris
+
+
+class EnhancedDataProvider(atheris.FuzzedDataProvider):
+    def ConsumeRandomBytes(self) -> bytes:
+        return self.ConsumeBytes(self.ConsumeIntInRange(0, self.remaining_bytes()))
+
+    def ConsumeRandomString(self) -> str:
+        return self.ConsumeUnicodeNoSurrogates(self.ConsumeIntInRange(0, self.remaining_bytes()))

fuzz/fuzz_form.py

@@ -46,6 +46,7 @@ dev = [
     "pytest-timeout==2.2.0",
     "ruff==0.3.0",
     "hatch",
+    "atheris==2.3.0; python_version != '3.12'",
 ]
 docs = [
     "mkdocs==1.5.3",

fuzz/fuzz_options_header.py

@@ -8,3 +8,4 @@ py==1.11.0
 pytest==8.0.2
 PyYAML==6.0.1
 ruff==0.3.0
+atheris==2.3.0