Add Atheris fuzz harnesses for URL, header, and decoder parsing

fuzz/README.md

@@ -0,0 +1,27 @@
+# Fuzzing HTTPX2
+
+Coverage-guided harnesses for the parts of HTTPX2 that consume untrusted
+server input: URL parsing, header parsing, and content-encoding decoders
+(gzip, deflate, brotli, zstd).
+
+Each `fuzz_*.py` module is an [Atheris](https://github.com/google/atheris)
+harness. The harnesses run continuously under
+[OSS-Fuzz](https://github.com/google/oss-fuzz); local runs require an Atheris
+install built against a libFuzzer-enabled Clang.
+
+## Local runs
+
+```shell
+uv pip install atheris
+uv run python fuzz/fuzz_urlparse.py -atheris_runs=100000
+uv run python fuzz/fuzz_headers.py  -atheris_runs=100000
+uv run python fuzz/fuzz_decoders.py -atheris_runs=100000
+```
+
+Pass a corpus directory as a positional argument to persist interesting
+inputs across runs:
+
+```shell
+mkdir -p fuzz/corpus_urlparse
+uv run python fuzz/fuzz_urlparse.py fuzz/corpus_urlparse
+```

fuzz/fuzz_decoders.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+import sys
+
+import atheris
+
+with atheris.instrument_imports():
+    from httpx2._decoders import (
+        BrotliDecoder,
+        DeflateDecoder,
+        GZipDecoder,
+        ZStandardDecoder,
+    )
+    from httpx2._exceptions import DecodingError
+
+DECODERS = [DeflateDecoder, GZipDecoder, BrotliDecoder, ZStandardDecoder]
+
+
+def TestOneInput(data: bytes) -> None:
+    if len(data) < 2:
+        return
+    fdp = atheris.FuzzedDataProvider(data)
+    decoder_cls = DECODERS[fdp.ConsumeIntInRange(0, len(DECODERS) - 1)]
+    try:
+        decoder = decoder_cls()
+    except ImportError:
+        return
+    # Feed the decoder several chunks to exercise streaming state.
+    num_chunks = fdp.ConsumeIntInRange(1, 8)
+    for _ in range(num_chunks):
+        chunk_size = fdp.ConsumeIntInRange(0, 4096)
+        chunk = fdp.ConsumeBytes(chunk_size)
+        try:
+            decoder.decode(chunk)
+        except DecodingError:
+            return
+    try:
+        decoder.flush()
+    except DecodingError:
+        return
+
+
+def main() -> None:
+    atheris.Setup(sys.argv, TestOneInput)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()

fuzz/fuzz_headers.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import sys
+
+import atheris
+
+with atheris.instrument_imports():
+    from httpx2 import Headers
+
+
+def TestOneInput(data: bytes) -> None:
+    fdp = atheris.FuzzedDataProvider(data)
+    num_headers = fdp.ConsumeIntInRange(0, 32)
+    raw: list[tuple[bytes, bytes]] = []
+    for _ in range(num_headers):
+        name = fdp.ConsumeBytes(fdp.ConsumeIntInRange(0, 64))
+        value = fdp.ConsumeBytes(fdp.ConsumeIntInRange(0, 256))
+        raw.append((name, value))
+    try:
+        headers = Headers(raw)
+    except (UnicodeDecodeError, ValueError):
+        return
+    # Exercise iteration and lookup paths.
+    for key in list(headers.keys()):
+        headers.get(key)
+    headers.multi_items()
+
+
+def main() -> None:
+    atheris.Setup(sys.argv, TestOneInput)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()

fuzz/fuzz_urlparse.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+import sys
+
+import atheris
+
+with atheris.instrument_imports():
+    from httpx2 import InvalidURL
+    from httpx2._urlparse import urlparse
+
+
+def TestOneInput(data: bytes) -> None:
+    try:
+        url = data.decode("utf-8", errors="replace")
+    except Exception:
+        return
+    try:
+        urlparse(url)
+    except (InvalidURL, UnicodeError, ValueError):
+        pass
+
+
+def main() -> None:
+    atheris.Setup(sys.argv, TestOneInput)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()