fuzzing: allow fuzzers to choose the serialization method (#556)

Signed-off-by: Adam Korczynski <[email protected]>

Author: AdamKorcz

tests/fuzzing/fuzz_sign_then_mutate_verify_with_valid_key.py

@@ -22,6 +22,7 @@
 import atheris  # type: ignore
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import ec
+from utils import _build_hashing_config_from_fdp
 from utils import any_files
 from utils import create_fuzz_files
 from utils import safe_write
@@ -98,8 +99,11 @@ def TestOneInput(data: bytes) -> None:
 
         key_spec = _pick_key_spec(fdp)
 
+        hcfg = _build_hashing_config_from_fdp(fdp)
+
         # Sign the directory root with a valid private key.
         scfg = model_signing.signing.Config()
+        scfg.set_hashing_config(hcfg)
         signer = scfg.use_elliptic_key_signer(private_key=key_spec["priv"])
         _ = signer.sign(model_path, sig_path)
 
@@ -134,6 +138,7 @@ def TestOneInput(data: bytes) -> None:
 
         # Verification must fail with ValueError because the tree changed.
         vcfg = model_signing.verifying.Config()
+        vcfg.set_hashing_config(hcfg)
         verifier = vcfg.use_elliptic_key_verifier(public_key=key_spec["pub"])
         try:
             verifier.verify(model_path, sig_path)

tests/fuzzing/fuzz_sign_verify_with_valid_key.py

@@ -23,6 +23,7 @@
 import atheris
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import ec
+from utils import _build_hashing_config_from_fdp
 from utils import any_files
 from utils import create_fuzz_files
 
@@ -98,11 +99,15 @@ def TestOneInput(data: bytes) -> None:
 
         key_spec = _pick_key_spec(fdp)
 
+        hcfg = _build_hashing_config_from_fdp(fdp)
+
         scfg = model_signing.signing.Config()
+        scfg.set_hashing_config(hcfg)
         signer = scfg.use_elliptic_key_signer(private_key=key_spec["priv"])
         _ = signer.sign(model_path, sig_path)
 
         vcfg = model_signing.verifying.Config()
+        vcfg.set_hashing_config(hcfg)
         verifier = vcfg.use_elliptic_key_verifier(public_key=key_spec["pub"])
         verifier.verify(model_path, sig_path)
 

tests/fuzzing/fuzz_sign_with_invalid_key.py

@@ -20,6 +20,7 @@
 # type: ignore
 import atheris
 from cryptography.hazmat.primitives import serialization
+from utils import _build_hashing_config_from_fdp
 from utils import any_files
 from utils import create_fuzz_files
 
@@ -63,7 +64,10 @@ def TestOneInput(data: bytes) -> None:
         model_path = str(root)
         sig_path = os.path.join(sigdir, "model.sig")
 
+        hcfg = _build_hashing_config_from_fdp(fdp)
+
         scfg = model_signing.signing.Config()
+        scfg.set_hashing_config(hcfg)
         signer = scfg.use_elliptic_key_signer(private_key=key_path)
         _ = signer.sign(model_path, sig_path)
 

tests/fuzzing/fuzz_sign_with_valid_key_verify_with_invalid_key.py

@@ -23,6 +23,7 @@
 import atheris
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import ec
+from utils import _build_hashing_config_from_fdp
 from utils import any_files
 from utils import create_fuzz_files
 
@@ -90,8 +91,11 @@ def TestOneInput(data: bytes) -> None:
         model_path = str(root)
         sig_path = os.path.join(sigdir, "model.sig")
 
+        hcfg = _build_hashing_config_from_fdp(fdp)
+
         # Sign using the valid private key created at module import time.
         scfg = model_signing.signing.Config()
+        scfg.set_hashing_config(hcfg)
         signer = scfg.use_elliptic_key_signer(private_key=_PRIV_PATH)
         _ = signer.sign(model_path, sig_path)
 
@@ -102,6 +106,7 @@ def TestOneInput(data: bytes) -> None:
 
         # Verify with the fuzzed public key.
         vcfg = model_signing.verifying.Config()
+        vcfg.set_hashing_config(hcfg)
         verifier = vcfg.use_elliptic_key_verifier(public_key=pubkey_path)
         verifier.verify(model_path, sig_path)
 

tests/fuzzing/fuzz_simple_sigstore.py

@@ -21,6 +21,7 @@
 # type: ignore
 import atheris
 from sigstore.models import TrustedRoot
+from utils import _build_hashing_config_from_fdp
 from utils import any_files
 from utils import create_fuzz_files
 
@@ -131,7 +132,10 @@ def TestOneInput(data: bytes) -> None:
             fdp.ConsumeBytes(64).decode("utf-8", errors="ignore") or "token"
         )
 
+        hcfg = _build_hashing_config_from_fdp(fdp)
+
         sc = signing.Config()
+        sc.set_hashing_config(hcfg)
         sc.use_sigstore_signer(
             use_staging=True, identity_token=sigstore_oidc_beacon_token
         )
@@ -142,6 +146,7 @@ def TestOneInput(data: bytes) -> None:
 
         # 7) Verify
         vc = verifying.Config()
+        vc.set_hashing_config(hcfg)
         vc.use_sigstore_verifier(
             identity=expected_identity,
             oidc_issuer=expected_oidc_issuer,

tests/fuzzing/fuzz_with_cert_chain.py

@@ -278,6 +278,23 @@ def key_to_pem(priv: rsa.RSAPrivateKey | ec.EllipticCurvePrivateKey) -> bytes:
     )
 
 
+def _build_hashing_config_from_fdp(
+    fdp: atheris.FuzzedDataProvider,
+    extra_ignores: list[Path],
+    signature_path: Path,
+) -> hashing.Config:
+    alg = ["sha256", "blake2", "blake3"][fdp.ConsumeIntInRange(0, 2)]
+    hcfg = hashing.Config().set_ignored_paths(
+        paths=[*list(extra_ignores), signature_path],
+        ignore_git_paths=fdp.ConsumeBool(),
+    )
+    if fdp.ConsumeBool():
+        hcfg.use_file_serialization(hashing_algorithm=alg)
+    else:
+        hcfg.use_shard_serialization(hashing_algorithm=alg)
+    return hcfg
+
+
 def TestOneInput(data: bytes):
     fdp = atheris.FuzzedDataProvider(data)
 
@@ -331,31 +348,23 @@ def TestOneInput(data: bytes):
     fname = f"signature-{_rand_utf8(fdp, 3, 12).replace('/', '_')}.sig"
     signature_path = model_path_p / fname
 
-    # Ignores
-    ignore_git = fdp.ConsumeBool()
+    # Ignores (collected for hashing config)
     extra_ignores: list[Path] = []
 
+    # Build hashing config (serialization + algorithm + ignores)
+    hcfg = _build_hashing_config_from_fdp(fdp, extra_ignores, signature_path)
+
     # 4) Sign and 5) Verify
     try:
         signing.Config().use_certificate_signer(
             private_key=leaf_key_path,
             signing_certificate=leaf_cert_path,
             certificate_chain=chain_paths,
-        ).set_hashing_config(
-            hashing.Config().set_ignored_paths(
-                paths=[*list(extra_ignores), signature_path],
-                ignore_git_paths=ignore_git,
-            )
-        ).sign(model_path_p, signature_path)
+        ).set_hashing_config(hcfg).sign(model_path_p, signature_path)
 
         verifying.Config().use_certificate_verifier(
             certificate_chain=chain_paths, log_fingerprints=False
-        ).set_hashing_config(
-            hashing.Config().set_ignored_paths(
-                paths=[*list(extra_ignores), signature_path],
-                ignore_git_paths=ignore_git,
-            )
-        ).verify(model_path_p, signature_path)
+        ).set_hashing_config(hcfg).verify(model_path_p, signature_path)
 
     finally:
         # Always clean up temp dirs

tests/fuzzing/utils.py

@@ -19,6 +19,8 @@
 # type: ignore
 import atheris
 
+import model_signing
+
 
 _SAFE_CHARS = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-"
 
@@ -126,3 +128,20 @@ def create_fuzz_files(root: Path, fdp: atheris.FuzzedDataProvider) -> int:
 def any_files(root: Path) -> bool:
     """True if there is at least one regular file under root."""
     return any(p.is_file() for p in root.rglob("*"))
+
+
+def _build_hashing_config_from_fdp(
+    fdp: atheris.FuzzedDataProvider,
+) -> "model_signing.hashing.Config":
+    """Randomize serialization strategy and hash algorithm."""
+    alg = ["sha256", "blake2", "blake3"][fdp.ConsumeIntInRange(0, 2)]
+
+    hcfg = model_signing.hashing.Config()
+    # Choose serialization mode: file vs shard
+    if fdp.ConsumeBool():
+        # File-based serialization.
+        hcfg.use_file_serialization(hashing_algorithm=alg)
+    else:
+        # Sharded file serialization
+        hcfg.use_shard_serialization(hashing_algorithm=alg)
+    return hcfg