Add OSS-Fuzz fuzz targets (experimental).

Author: aaugustin

fuzzing/fuzz_http11_request_parser.py

@@ -0,0 +1,35 @@
+import sys
+
+import atheris
+
+
+with atheris.instrument_imports():
+    from websockets.exceptions import SecurityError
+    from websockets.http11 import Request
+    from websockets.streams import StreamReader
+
+
+def test_one_input(data):
+    reader = StreamReader()
+    reader.feed_data(data)
+    reader.feed_eof()
+
+    try:
+        Request.parse(
+            reader.read_line,
+        )
+    except (
+        EOFError,  # connection is closed without a full HTTP request
+        SecurityError,  # request exceeds a security limit
+        ValueError,  # request isn't well formatted
+    ):
+        pass
+
+
+def main():
+    atheris.Setup(sys.argv, test_one_input)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()

fuzzing/fuzz_http11_response_parser.py

@@ -0,0 +1,38 @@
+import sys
+
+import atheris
+
+
+with atheris.instrument_imports():
+    from websockets.exceptions import SecurityError
+    from websockets.http11 import Response
+    from websockets.streams import StreamReader
+
+
+def test_one_input(data):
+    reader = StreamReader()
+    reader.feed_data(data)
+    reader.feed_eof()
+
+    try:
+        Response.parse(
+            reader.read_line,
+            reader.read_exact,
+            reader.read_to_eof,
+        )
+    except (
+        EOFError,  # connection is closed without a full HTTP response.
+        SecurityError,  # response exceeds a security limit.
+        LookupError,  # response isn't well formatted.
+        ValueError,  # response isn't well formatted.
+    ):
+        pass
+
+
+def main():
+    atheris.Setup(sys.argv, test_one_input)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()