Add typing and docstrings to all functions and classes

Author: cmeister2

.flake8

@@ -0,0 +1,6 @@
+[flake8]
+ignore =
+    D400
+    D401
+    E203
+max-line-length = 88

.gitignore

@@ -67,4 +67,5 @@ missing
 
 # Python
 .venv/
-__pycache__/
+__pycache__/
+logs/

poetry.lock

@@ -21,11 +21,18 @@ black = "^25.1.0"
 flake8 = "^7.1.2"
 mypy = "^1.15.0"
 flake8-isort = "^6.1.2"
+flake8-black = "^0.3.6"
+flake8-docstrings = "^1.7.0"
 
 [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"
+read_corpus = "curl_fuzzer_tools.read_corpus:run"
+generate_corpus = "curl_fuzzer_tools.generate_corpus:run"
+corpus_to_pcap = "curl_fuzzer_tools.corpus_to_pcap:run"
+
+[tool.mypy]
+warn_unused_configs = true
+disallow_untyped_defs = true
+warn_return_any = true

pyproject.toml

@@ -0,0 +1,6 @@
+"""Tooling for the curl-fuzzer repository."""
+
+from .logger import common_logging
+
+# Import * imports
+__all__ = ["common_logging"]

src/curl_fuzzer_tools/__init__.py

@@ -1,13 +1,18 @@
-#!/usr/bin/env python
-#
-# Common corpus functions
+"""Common corpus functions"""
+
 import logging
 import struct
+from pathlib import Path
+from typing import BinaryIO, Optional
+
+from curl_fuzzer_tools.curl_test_data import TestData
 
 log = logging.getLogger(__name__)
 
 
 class BaseType(object):
+    """Known TLV types"""
+
     TYPE_URL = 1
     TYPE_RSP0 = 2
     TYPE_USERNAME = 3
@@ -59,7 +64,7 @@ class BaseType(object):
     TYPE_WS_OPTIONS = 49
     TYPE_CONNECT_ONLY = 50
     TYPE_HSTS = 51
-    TYPE_HTTPPOSTBODY = 52 # https://curl.se/libcurl/c/CURLOPT_HTTPPOST.html
+    TYPE_HTTPPOSTBODY = 52  # https://curl.se/libcurl/c/CURLOPT_HTTPPOST.html
 
     TYPEMAP = {
         TYPE_URL: "CURLOPT_URL",
@@ -118,56 +123,79 @@ class BaseType(object):
 
 
 class TLVEncoder(BaseType):
-    def __init__(self, output, test_data):
+    """Class for encoding TLVs"""
+
+    def __init__(self, output: BinaryIO, test_data: TestData) -> None:
+        """Create a TLVEncoder object"""
         self.output = output
         self.test_data = test_data
 
-    def write_string(self, tlv_type, wstring):
+    def write_string(self, tlv_type: int, wstring: str) -> None:
+        """Write a string TLV to the output"""
         data = wstring.encode("utf-8")
         self.write_tlv(tlv_type, len(data), data)
 
-    def write_u32(self, tlv_type, num):
+    def write_u32(self, tlv_type: int, num: int) -> None:
+        """Write an unsigned 32-bit integer TLV to the output"""
         data = struct.pack("!L", num)
         self.write_tlv(tlv_type, len(data), data)
 
-    def write_bytes(self, tlv_type, bytedata):
+    def write_bytes(self, tlv_type: int, bytedata: bytes) -> None:
+        """Write a bytes TLV to the output"""
         self.write_tlv(tlv_type, len(bytedata), bytedata)
 
-    def maybe_write_string(self, tlv_type, wstring):
+    def maybe_write_string(self, tlv_type: int, wstring: Optional[str]) -> None:
+        """Write a string TLV to the output if specified"""
         if wstring is not None:
             self.write_string(tlv_type, wstring)
 
-    def maybe_write_u32(self, tlv_type, num):
+    def maybe_write_u32(self, tlv_type: int, num: Optional[int]) -> None:
+        """Write an unsigned 32-bit integer TLV to the output if specified"""
         if num is not None:
             self.write_u32(tlv_type, num)
 
-    def maybe_write_response(self, rsp_type, rsp, rsp_file, rsp_test):
+    def maybe_write_response(
+        self, rsp_type: int, rsp: Optional[str], rsp_file: Optional[Path], rsp_test: int
+    ) -> None:
+        """Write a response TLV to the output if specified"""
         if rsp:
-            self.write_bytes(rsp_type,
-                             rsp.encode("utf-8"))
+            self.write_bytes(rsp_type, rsp.encode("utf-8"))
         elif rsp_file:
             with open(rsp_file, "rb") as g:
                 self.write_bytes(rsp_type, g.read())
         elif rsp_test:
             wstring = self.test_data.get_test_data(rsp_test)
             self.write_bytes(rsp_type, wstring.encode("utf-8"))
 
-    def write_mimepart(self, namevalue):
+    def write_mimepart(self, namevalue: str) -> None:
+        """Write a MIME part TLV to the output"""
         (name, value) = namevalue.split(":", 1)
 
         # Create some mimepart TLVs for the name and value
-        name_tlv = self.encode_tlv(self.TYPE_MIME_PART_NAME, len(name), name)
-        value_tlv = self.encode_tlv(self.TYPE_MIME_PART_DATA, len(value), value)
+        name_bytes = name.encode("utf-8")
+        value_bytes = value.encode("utf-8")
+
+        name_tlv = self.encode_tlv(
+            self.TYPE_MIME_PART_NAME, len(name_bytes), name_bytes
+        )
+        value_tlv = self.encode_tlv(
+            self.TYPE_MIME_PART_DATA, len(value_bytes), value_bytes
+        )
 
         # Combine the two TLVs into a single TLV.
         part_tlv = name_tlv + value_tlv
         self.write_tlv(self.TYPE_MIME_PART, len(part_tlv), part_tlv)
 
-    def encode_tlv(self, tlv_type, tlv_length, tlv_data=None):
-        log.debug("Encoding TLV %r, length %d, data %r",
-                  self.TYPEMAP.get(tlv_type, "<unknown>"),
-                  tlv_length,
-                  tlv_data)
+    def encode_tlv(
+        self, tlv_type: int, tlv_length: int, tlv_data: Optional[bytes] = None
+    ) -> bytes:
+        """Encodes the Type, Length, and Value into a bytes array"""
+        log.debug(
+            "Encoding TLV %r, length %d, data %r",
+            self.TYPEMAP.get(tlv_type, "<unknown>"),
+            tlv_length,
+            tlv_data,
+        )
 
         data = struct.pack("!H", tlv_type)
         data = data + struct.pack("!L", tlv_length)
@@ -176,56 +204,79 @@ def encode_tlv(self, tlv_type, tlv_length, tlv_data=None):
 
         return data
 
-    def write_tlv(self, tlv_type, tlv_length, tlv_data=None):
-        log.debug("Writing TLV %r, length %d, data %r",
-                  self.TYPEMAP.get(tlv_type, "<unknown>"),
-                  tlv_length,
-                  tlv_data)
+    def write_tlv(
+        self, tlv_type: int, tlv_length: int, tlv_data: Optional[bytes] = None
+    ) -> None:
+        """Writes an encoded TLV to the output as bytes"""
+        log.debug(
+            "Writing TLV %r, length %d, data %r",
+            self.TYPEMAP.get(tlv_type, "<unknown>"),
+            tlv_length,
+            tlv_data,
+        )
 
         data = self.encode_tlv(tlv_type, tlv_length, tlv_data)
         self.output.write(data)
 
 
+class TLVContents(BaseType):
+    """Class for TLV contents"""
+
+    TLV_DECODE_FMT = "!HL"
+    TLV_DECODE_FMT_LEN = struct.calcsize(TLV_DECODE_FMT)
+
+    def __init__(self, data: bytes) -> None:
+        """Create a TLVContents object"""
+        # Parse the data to populate the TLV fields
+        (stype, slen) = struct.unpack(
+            self.TLV_DECODE_FMT, data[0 : self.TLV_DECODE_FMT_LEN]
+        )
+        self.type = int(stype)
+        self.length = int(slen)
+
+        # Get the remaining data and store it.
+        self.data = data[
+            self.TLV_DECODE_FMT_LEN : self.TLV_DECODE_FMT_LEN + self.length
+        ]
+
+    def __repr__(self) -> str:
+        """Return a string representation of the TLVContents object"""
+        stype = self.TYPEMAP.get(self.type, "<unknown>")
+        return (
+            f"{self.__class__.__name__}(type={stype!r} ({self.type!r}), "
+            f"length={self.length!r}, data={self.data!r})"
+        )
+
+    def total_length(self) -> int:
+        """Return the total length of the TLV, including the header"""
+        return self.TLV_DECODE_FMT_LEN + self.length
+
+
 class TLVDecoder(BaseType):
-    def __init__(self, inputdata):
+    """Class for decoding TLVs"""
+
+    def __init__(self, inputdata: bytes) -> None:
+        """Create a TLVDecoder object"""
         self.inputdata = inputdata
         self.pos = 0
-        self.tlv = None
+        self.tlv: Optional["TLVContents"] = None
 
-    def __iter__(self):
+    def __iter__(self) -> "TLVDecoder":
+        """Return an iterator for the TLVs"""
         self.pos = 0
         self.tlv = None
         return self
 
-    def __next__(self):
+    def __next__(self) -> "TLVContents":
+        """Return the next TLV in the input data"""
         if self.tlv:
             self.pos += self.tlv.total_length()
 
-        if (self.pos + TLVHeader.TLV_DECODE_FMT_LEN) > len(self.inputdata):
+        if (self.pos + TLVContents.TLV_DECODE_FMT_LEN) > len(self.inputdata):
             raise StopIteration
 
         # Get the next TLV
-        self.tlv = TLVHeader(self.inputdata[self.pos:])
+        self.tlv = TLVContents(self.inputdata[self.pos :])
         return self.tlv
 
     next = __next__
-
-
-class TLVHeader(BaseType):
-    TLV_DECODE_FMT = "!HL"
-    TLV_DECODE_FMT_LEN = struct.calcsize(TLV_DECODE_FMT)
-
-    def __init__(self, data):
-        # Parse the data to populate the TLV fields
-        (self.type, self.length) = struct.unpack(self.TLV_DECODE_FMT, data[0:self.TLV_DECODE_FMT_LEN])
-
-        # Get the remaining data and store it.
-        self.data = data[self.TLV_DECODE_FMT_LEN:self.TLV_DECODE_FMT_LEN + self.length]
-
-    def __repr__(self):
-        return ("{self.__class__.__name__}(type={stype!r} ({self.type!r}), length={self.length!r}, data={self.data!r})"
-                .format(self=self,
-                        stype=self.TYPEMAP.get(self.type, "<unknown>")))
-
-    def total_length(self):
-        return self.TLV_DECODE_FMT_LEN + self.length