Move Python files into tooling directory and create Poetry configuration

Author: cmeister2

.gitignore

@@ -64,3 +64,7 @@ missing
 /curl_fuzzer_smtp_seed_corpus.zip
 /curl_fuzzer_tftp
 /curl_fuzzer_tftp_seed_corpus.zip
+
+# Python
+.venv/
+__pycache__/

poetry.lock

@@ -0,0 +1,31 @@
+[project]
+name = "curl-fuzzer-tools"
+version = "0.1.0"
+description = "Tooling for the curl-fuzzer repository"
+authors = [
+    {name = "Max Dymond",email = "[email protected]"}
+]
+readme = "README.md"
+requires-python = "^3.9"
+dependencies = [
+    "scapy (>=2.6.1,<3.0.0)"
+]
+
+
+[build-system]
+requires = ["poetry-core>=2.0.0,<3.0.0"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.poetry.group.dev.dependencies]
+black = "^25.1.0"
+flake8 = "^7.1.2"
+mypy = "^1.15.0"
+flake8-isort = "^6.1.2"
+
+[tool.isort]
+profile = "black"
+
+[tool.poetry.scripts]
+read_corpus = "curl_fuzzer_tools.read_corpus:main"
+generate_corpus = "curl_fuzzer_tools.generate_corpus:main"
+corpus_to_pcap = "curl_fuzzer_tools.corpus_to_pcap:main"

pyproject.toml

@@ -3,6 +3,7 @@
 # Common corpus functions
 import logging
 import struct
+
 log = logging.getLogger(__name__)
 
 

src/curl_fuzzer_tools/__init__.py

@@ -4,6 +4,7 @@
 
 from enum import Enum
 
+
 class CurlOptHttpAuth(Enum):
     #define CURLAUTH_NONE         ((unsigned long)0)
     CURLAUTH_NONE = 0

corpus.py renamed to src/curl_fuzzer_tools/corpus.py

@@ -5,34 +5,38 @@
 import argparse
 import logging
 import sys
-from scapy.all import IP, TCP, wrpcap
-import corpus
+
+from scapy.all import wrpcap
+from scapy.layers.inet import IP, TCP
+
+from curl_fuzzer_tools.corpus import BaseType, TLVDecoder
+
 log = logging.getLogger(__name__)
 
 
 # All the responses we want to convert to pcap
 RESPONSES = [
-    corpus.BaseType.TYPE_RSP0,
-    corpus.BaseType.TYPE_RSP1,
-    corpus.BaseType.TYPE_RSP2,
-    corpus.BaseType.TYPE_RSP3,
-    corpus.BaseType.TYPE_RSP4,
-    corpus.BaseType.TYPE_RSP5,
-    corpus.BaseType.TYPE_RSP6,
-    corpus.BaseType.TYPE_RSP7,
-    corpus.BaseType.TYPE_RSP8,
-    corpus.BaseType.TYPE_RSP9,
-    corpus.BaseType.TYPE_RSP10,
-    corpus.BaseType.TYPE_SECRSP0,
-    corpus.BaseType.TYPE_SECRSP1,
+    BaseType.TYPE_RSP0,
+    BaseType.TYPE_RSP1,
+    BaseType.TYPE_RSP2,
+    BaseType.TYPE_RSP3,
+    BaseType.TYPE_RSP4,
+    BaseType.TYPE_RSP5,
+    BaseType.TYPE_RSP6,
+    BaseType.TYPE_RSP7,
+    BaseType.TYPE_RSP8,
+    BaseType.TYPE_RSP9,
+    BaseType.TYPE_RSP10,
+    BaseType.TYPE_SECRSP0,
+    BaseType.TYPE_SECRSP1,
 ]
 
 
 def corpus_to_pcap(options):
     response_tlvs = {}
 
     with open(options.input, "rb") as f:
-        dec = corpus.TLVDecoder(f.read())
+        dec = TLVDecoder(f.read())
         for tlv in dec:
             if tlv.type in RESPONSES:
                 log.debug("Found response: %s", tlv)
@@ -49,7 +53,7 @@ def corpus_to_pcap(options):
             # By default generate a packet with source port 80. This hints at HTTP; in future we can be smart and
             # pick a port that'll influence Wireshark. But for now, you can just Decode As.. in Wireshark to get
             # whatever protocol you want.
-            pkt = IP() / TCP(sport=80, flags='SA') / tlv.data
+            pkt = IP() / TCP(sport=80, flags="SA") / tlv.data
             log.debug("Converted %s to packet: %s", tlv.TYPEMAP[rsp], pkt)
             response_packets.append(pkt)
 
@@ -81,6 +85,7 @@ def setup_logging():
 
 class ScriptRC(object):
     """Enum for script return codes"""
+
     SUCCESS = 0
     FAILURE = 1
     EXCEPTION = 2
@@ -107,5 +112,5 @@ def main():
     return rc
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     sys.exit(main())

corpus_curl_opt_http_auth.py renamed to src/curl_fuzzer_tools/corpus_curl_opt_http_auth.py

@@ -21,11 +21,11 @@
 #
 """Module for extracting test data from the test data folder"""
 
-from __future__ import (absolute_import, division, print_function,
-                        unicode_literals)
+from __future__ import absolute_import, division, print_function, unicode_literals
+
+import logging
 import os
 import re
-import logging
 
 log = logging.getLogger(__name__)
 

corpus_to_pcap.py renamed to src/curl_fuzzer_tools/corpus_to_pcap.py

@@ -6,40 +6,67 @@
 import logging
 import os
 import sys
-import corpus
-from corpus_curl_opt_http_auth import CurlOptHttpAuth
+
+from curl_fuzzer_tools.corpus import TLVEncoder
+from curl_fuzzer_tools.corpus_curl_opt_http_auth import CurlOptHttpAuth
+from curl_fuzzer_tools.curl_test_data import TestData
+
 log = logging.getLogger(__name__)
 
 
 def generate_corpus(options):
     sys.path.append(options.curl_test_dir)
-    import curl_test_data
 
-    td = curl_test_data.TestData(os.path.join(options.curl_test_dir,
-                                              "data"))
+    td = TestData(os.path.join(options.curl_test_dir, "data"))
 
     with open(options.output, "wb") as f:
-        enc = corpus.TLVEncoder(f, td)
+        enc = TLVEncoder(f, td)
 
         # Write the URL to the file.
         enc.write_string(enc.TYPE_URL, options.url)
 
         # Write any responses to the file.
-        enc.maybe_write_response(enc.TYPE_RSP0, options.rsp0, options.rsp0file, options.rsp0test)
-        enc.maybe_write_response(enc.TYPE_RSP1, options.rsp1, options.rsp1file, options.rsp1test)
-        enc.maybe_write_response(enc.TYPE_RSP2, options.rsp2, options.rsp2file, options.rsp2test)
-        enc.maybe_write_response(enc.TYPE_RSP3, options.rsp3, options.rsp3file, options.rsp3test)
-        enc.maybe_write_response(enc.TYPE_RSP4, options.rsp4, options.rsp4file, options.rsp4test)
-        enc.maybe_write_response(enc.TYPE_RSP5, options.rsp5, options.rsp5file, options.rsp5test)
-        enc.maybe_write_response(enc.TYPE_RSP6, options.rsp6, options.rsp6file, options.rsp6test)
-        enc.maybe_write_response(enc.TYPE_RSP7, options.rsp7, options.rsp7file, options.rsp7test)
-        enc.maybe_write_response(enc.TYPE_RSP8, options.rsp8, options.rsp8file, options.rsp8test)
-        enc.maybe_write_response(enc.TYPE_RSP9, options.rsp9, options.rsp9file, options.rsp9test)
-        enc.maybe_write_response(enc.TYPE_RSP10, options.rsp10, options.rsp10file, options.rsp10test)
+        enc.maybe_write_response(
+            enc.TYPE_RSP0, options.rsp0, options.rsp0file, options.rsp0test
+        )
+        enc.maybe_write_response(
+            enc.TYPE_RSP1, options.rsp1, options.rsp1file, options.rsp1test
+        )
+        enc.maybe_write_response(
+            enc.TYPE_RSP2, options.rsp2, options.rsp2file, options.rsp2test
+        )
+        enc.maybe_write_response(
+            enc.TYPE_RSP3, options.rsp3, options.rsp3file, options.rsp3test
+        )
+        enc.maybe_write_response(
+            enc.TYPE_RSP4, options.rsp4, options.rsp4file, options.rsp4test
+        )
+        enc.maybe_write_response(
+            enc.TYPE_RSP5, options.rsp5, options.rsp5file, options.rsp5test
+        )
+        enc.maybe_write_response(
+            enc.TYPE_RSP6, options.rsp6, options.rsp6file, options.rsp6test
+        )
+        enc.maybe_write_response(
+            enc.TYPE_RSP7, options.rsp7, options.rsp7file, options.rsp7test
+        )
+        enc.maybe_write_response(
+            enc.TYPE_RSP8, options.rsp8, options.rsp8file, options.rsp8test
+        )
+        enc.maybe_write_response(
+            enc.TYPE_RSP9, options.rsp9, options.rsp9file, options.rsp9test
+        )
+        enc.maybe_write_response(
+            enc.TYPE_RSP10, options.rsp10, options.rsp10file, options.rsp10test
+        )
 
         # Write any second socket responses to the file.
-        enc.maybe_write_response(enc.TYPE_SECRSP0, options.secrsp0, options.secrsp0file, options.secrsp0test)
-        enc.maybe_write_response(enc.TYPE_SECRSP1, options.secrsp1, options.secrsp1file, options.secrsp1test)
+        enc.maybe_write_response(
+            enc.TYPE_SECRSP0, options.secrsp0, options.secrsp0file, options.secrsp0test
+        )
+        enc.maybe_write_response(
+            enc.TYPE_SECRSP1, options.secrsp1, options.secrsp1file, options.secrsp1test
+        )
 
         # Write other options to file.
         enc.maybe_write_string(enc.TYPE_USERNAME, options.username)
@@ -59,7 +86,9 @@ def generate_corpus(options):
         enc.maybe_write_string(enc.TYPE_XOAUTH2_BEARER, options.bearertoken)
         enc.maybe_write_string(enc.TYPE_USERPWD, options.user_and_pass)
         enc.maybe_write_string(enc.TYPE_USERAGENT, options.useragent)
-        enc.maybe_write_string(enc.TYPE_SSH_HOST_PUBLIC_KEY_SHA256, options.hostpksha256)
+        enc.maybe_write_string(
+            enc.TYPE_SSH_HOST_PUBLIC_KEY_SHA256, options.hostpksha256
+        )
         enc.maybe_write_string(enc.TYPE_HSTS, options.hsts)
 
         enc.maybe_write_u32(enc.TYPE_OPTHEADER, options.optheader)
@@ -75,8 +104,10 @@ def generate_corpus(options):
         if options.httpauth:
             # translate a string HTTP auth name to an unsigned long bitmask
             # value in the format CURLOPT_HTTPAUTH expects
-            log.debug(f"Mapping provided CURLOPT_HTTPAUTH='{options.httpauth}' "
-                f"to {CurlOptHttpAuth[options.httpauth].value}L (ulong)")
+            log.debug(
+                f"Mapping provided CURLOPT_HTTPAUTH='{options.httpauth}' "
+                f"to {CurlOptHttpAuth[options.httpauth].value}L (ulong)"
+            )
             http_auth_value = CurlOptHttpAuth[options.httpauth].value
             enc.maybe_write_u32(enc.TYPE_HTTPAUTH, http_auth_value)
 
@@ -150,9 +181,9 @@ def get_options():
     parser.add_argument("--useragent", type=str)
     parser.add_argument("--netrclevel", type=int)
     parser.add_argument("--hostpksha256", type=str)
-    parser.add_argument("--wsoptions", action='store_true')
+    parser.add_argument("--wsoptions", action="store_true")
     parser.add_argument("--connectonly", type=int)
-    parser.add_argument("--post", action='store_true')
+    parser.add_argument("--post", action="store_true")
     parser.add_argument("--hsts")
 
     upload1 = parser.add_mutually_exclusive_group()
@@ -189,6 +220,7 @@ def setup_logging():
 
 class ScriptRC(object):
     """Enum for script return codes"""
+
     SUCCESS = 0
     FAILURE = 1
     EXCEPTION = 2
@@ -215,5 +247,5 @@ def main():
     return rc
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     sys.exit(main())

curl_test_data.py renamed to src/curl_fuzzer_tools/curl_test_data.py

@@ -5,13 +5,15 @@
 import argparse
 import logging
 import sys
-import corpus
+
+from curl_fuzzer_tools.corpus import TLVDecoder
+
 log = logging.getLogger(__name__)
 
 
 def read_corpus(options):
     with open(options.input, "rb") as f:
-        dec = corpus.TLVDecoder(f.read())
+        dec = TLVDecoder(f.read())
         for tlv in dec:
             print(tlv)
 
@@ -39,6 +41,7 @@ def setup_logging():
 
 class ScriptRC(object):
     """Enum for script return codes"""
+
     SUCCESS = 0
     FAILURE = 1
     EXCEPTION = 2
@@ -65,5 +68,5 @@ def main():
     return rc
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     sys.exit(main())