fix(test): reject mutation-harness false positives and honor fixture trust

.github/workflows/ci.yml

@@ -151,8 +151,13 @@ jobs:
       - name: Make binary executable
         run: chmod +x target/release/examples/verify_bundle
 
-      - name: Clone sigstore-conformance test data
-        run: git clone --depth 1 https://github.com/sigstore/sigstore-conformance.git
+      - name: Test mutation harness
+        run: python3 -m unittest discover -s tests -p test_mutation_fuzzer.py
 
-      - name: Run mutation fuzzer
-        run: python3 tests/mutation_fuzzer.py --verifier-type rust
+      - name: Mutate checked-in message signature and DSSE fixtures offline
+        run: |
+          bundles=crates/sigstore-verify/test_data/bundles
+          roots=crates/sigstore-trust-root/src
+          python3 tests/mutation_fuzzer.py --bundle "$bundles/cosign-v3-blob.sigstore.json" --artifact "$bundles/cosign-v3-blob.txt" --trust-root "$roots/trusted_root.json"
+          python3 tests/mutation_fuzzer.py --bundle "$bundles/conda-attestation.sigstore.json" --artifact "$bundles/signed-package-2.1.0-hb0f4dca_0.conda" --trust-root "$roots/trusted_root.json"
+          python3 tests/mutation_fuzzer.py --bundle "$bundles/conda-attestation-rekor2.sigstore.json" --artifact "$bundles/signed-package-2.1.0-hb0f4dca_0.conda" --trust-root "$roots/trusted_root_staging.json"

crates/sigstore-verify/examples/verify_bundle.rs

@@ -62,6 +62,8 @@ async fn main() {
     let mut identity_regexp: Option<String> = None;
     let mut issuer: Option<String> = None;
     let mut instance: Option<String> = None;
+    let mut trusted_root_path: Option<String> = None;
+    let mut tuf_root_path: Option<String> = None;
     let mut staging = false;
     let mut positional: Vec<String> = Vec::new();
 
@@ -100,6 +102,19 @@ async fn main() {
                 }
                 instance = Some(args[i].clone());
             }
+            "--trusted-root" | "--tuf-root" => {
+                let option = args[i].clone();
+                i += 1;
+                let value = args.get(i).cloned().unwrap_or_else(|| {
+                    eprintln!("Error: {option} requires a file path");
+                    process::exit(2);
+                });
+                if option == "--trusted-root" {
+                    trusted_root_path = Some(value);
+                } else {
+                    tuf_root_path = Some(value);
+                }
+            }
             "--staging" => {
                 staging = true;
             }
@@ -125,6 +140,18 @@ async fn main() {
         process::exit(1);
     }
 
+    if usize::from(instance.is_some())
+        + usize::from(trusted_root_path.is_some())
+        + usize::from(staging)
+        > 1
+        || (instance.is_some() != tuf_root_path.is_some())
+    {
+        eprintln!(
+            "Error: select only one of --trusted-root, --staging, or --instance with --tuf-root"
+        );
+        process::exit(2);
+    }
+
     let artifact_or_digest = &positional[0];
     let bundle_path = &positional[1];
 
@@ -150,11 +177,23 @@ async fn main() {
     };
 
     // Load trusted root (staging or production Sigstore instance)
-    let trusted_root = if let Some(url) = instance {
+    let trusted_root = if let Some(path) = trusted_root_path {
+        let root = fs::read_to_string(path)
+            .map_err(|e| e.to_string())
+            .and_then(|json| TrustedRoot::from_json(&json).map_err(|e| e.to_string()));
+        root.unwrap_or_else(|e| {
+            eprintln!("Error loading trusted root: {e}");
+            process::exit(2);
+        })
+    } else if let Some(url) = instance {
         println!("  Using: custom instance ({})", url);
+        let bootstrap = fs::read(tuf_root_path.unwrap()).unwrap_or_else(|e| {
+            eprintln!("Error reading trusted TUF bootstrap: {e}");
+            process::exit(2);
+        });
         let config = sigstore_trust_root::tuf::TufConfig::custom(
             &url,
-            sigstore_trust_root::tuf::TufBootstrap::UnsafeCachedRoot,
+            sigstore_trust_root::tuf::TufBootstrap::trusted(bootstrap),
         );
         match TrustedRoot::from_tuf(config).await {
             Ok(root) => root,
@@ -294,7 +333,8 @@ fn print_usage(program: &str) {
     eprintln!("  --certificate-identity <ID>        Required certificate identity (exact match)");
     eprintln!("  --certificate-identity-regexp <RE> Required certificate identity (regex)");
     eprintln!("  --certificate-oidc-issuer <ISSUER> Required OIDC issuer");
-    eprintln!("  --instance <URL>                   Use a custom Sigstore instance");
+    eprintln!("  --trusted-root <FILE>              Use a local Sigstore trusted root (offline)");
+    eprintln!("  --instance <URL> --tuf-root <FILE>  Custom instance with trusted TUF bootstrap");
     eprintln!("  --staging                          Use Sigstore staging instance");
     eprintln!("  -h, --help                         Print this help message");
     eprintln!();

tests/mutation_fuzzer.py

@@ -820,11 +820,23 @@ def get_mutations_for_bundle(bundle: dict) -> list[Mutation]:
         "invalid_base64_certificate",  # Only targets single certificate
     }
 
+    entries = bundle.get("verificationMaterial", {}).get("tlogEntries", [])
+    kind_version = entries[0].get("kindVersion", {}) if entries else {}
+    rekor_v2 = kind_version.get("version") == "0.0.2" and kind_version.get("kind") in {"hashedrekord", "dsse"}
+    # Rekor v2 authenticates tree size/root in the checkpoint, not the duplicate
+    # proof fields. Log IDs are hints, and integratedTime is not signed in v2.
+    v1_authenticated_only = {
+        "inclusion_proof_wrong_root_hash", "inclusion_proof_wrong_tree_size",
+        "wrong_log_id", "integrated_time_future", "integrated_time_ancient",
+        "integrated_time_zero", "integrated_time_negative",
+    }
     applicable = []
     for mutation in ALL_MUTATIONS:
         name = mutation.name
 
         # Filter out mutations that don't apply to this bundle type
+        if rekor_v2 and name in v1_authenticated_only:
+            continue
         if name in dsse_only_mutations and not has_dsse:
             continue
         if name in message_sig_only_mutations and not has_message_sig:
@@ -978,7 +990,21 @@ def verify_bundle(self, bundle_path: Path, artifact_path: Path,
                 text=True,
                 timeout=30
             )
-            return (result.returncode == 0, result.stdout + result.stderr)
+            output = result.stdout + result.stderr
+            if result.returncode == 0:
+                return (True, output)
+            if result.returncode != 1 or any(marker in output for marker in (
+                "panicked at", "Traceback (most recent call last)", "fatal error:",
+            )):
+                raise RuntimeError(f"Verifier crashed or failed to run (exit {result.returncode}): {output[:500]}")
+            rejection_markers = {
+                VerifierType.SIGSTORE_RUST: ("Error parsing bundle:", "Verification error:", "Verification: FAILED"),
+                VerifierType.COSIGN: ("verification failed", "no matching signatures", "invalid bundle"),
+                VerifierType.SIGSTORE_PYTHON: ("Verification failed", "VerificationFailure", "InvalidBundle"),
+            }
+            if not any(marker in output for marker in rejection_markers[self.verifier_type]):
+                raise RuntimeError(f"Unrecognized verifier failure, not a verified rejection: {output[:500]}")
+            return (False, output)
         except subprocess.TimeoutExpired:
             raise RuntimeError("Verifier timed out")
         except FileNotFoundError:
@@ -997,6 +1023,8 @@ def _build_verify_command(self, bundle_path: Path, artifact_path: Path,
         if self.verifier_type == VerifierType.SIGSTORE_RUST:
             # Our verify_bundle example
             cmd.extend(["--certificate-identity-regexp", ".*"])
+            if trust_root_path:
+                cmd.extend(["--trusted-root", str(trust_root_path)])
             cmd.append(str(artifact_path))
             cmd.append(str(bundle_path))
 
@@ -1073,6 +1101,9 @@ def fuzz_bundle(
         Returns:
             List of (correctly_rejected, mutation_name, details) tuples
         """
+        success, output = self.verify_bundle(bundle_path, artifact_path, trust_root_path)
+        if not success:
+            raise RuntimeError(f"Unmodified baseline did not verify: {bundle_path}\n{output[:500]}")
         bundle = self.load_bundle(bundle_path)
 
         # Get mutations applicable to this bundle type
@@ -1085,6 +1116,8 @@ def fuzz_bundle(
             # Filter provided mutations to only those applicable to this bundle
             mutations = [m for m in mutations if m.name in applicable_names]
 
+        if not mutations:
+            raise RuntimeError(f"No applicable mutations for {bundle_path}")
         results = []
         for mutation in mutations:
             result = self.run_mutation_test(
@@ -1271,4 +1304,8 @@ def main():
 
 
 if __name__ == "__main__":
-    exit(main())
+    try:
+        sys.exit(main())
+    except (RuntimeError, OSError, ValueError) as error:
+        print(f"Harness error: {error}", file=sys.stderr)
+        sys.exit(2)

tests/test_mutation_fuzzer.py

@@ -0,0 +1,62 @@
+"""Run with: python3 -m unittest discover -s tests -p test_mutation_fuzzer.py."""
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from mutation_fuzzer import BundleMutationFuzzer, get_mutations_for_bundle
+
+
+class HarnessChecks(unittest.TestCase):
+    def test_baseline_trust_root_and_failure_classification(self):
+        fuzzer = BundleMutationFuzzer(verifier_command=[sys.executable])
+        bundle, artifact, root = map(Path, ["bundle.json", "artifact", "root.json"])
+        command = fuzzer._build_verify_command(bundle, artifact, root)
+        self.assertEqual(command[command.index("--trusted-root") + 1], str(root))
+        for code, output, expected in [
+            (0, "Verification: SUCCESS", True),
+            (1, "Verification error: invalid signature", False),
+            (1, "Error parsing bundle: invalid JSON", False),
+            (1, "could not load trust root", None),
+            (1, "", None),
+            (2, "usage error", None),
+            (101, "panicked at", None),
+            (-11, "", None),
+        ]:
+            with self.subTest(code=code, output=output), patch("mutation_fuzzer.subprocess.run", return_value=subprocess.CompletedProcess(command, code, output, "")):
+                if expected is None:
+                    with self.assertRaises(RuntimeError):
+                        fuzzer.verify_bundle(bundle, artifact, root)
+                else:
+                    self.assertEqual(fuzzer.verify_bundle(bundle, artifact, root)[0], expected)
+        with patch.object(fuzzer, "verify_bundle", return_value=(False, "invalid signature")):
+            with self.assertRaisesRegex(RuntimeError, "baseline"):
+                fuzzer.fuzz_bundle(bundle, artifact, root)
+        with patch("mutation_fuzzer.subprocess.run", side_effect=subprocess.TimeoutExpired(command, 30)):
+            with self.assertRaisesRegex(RuntimeError, "timed out"):
+                fuzzer.verify_bundle(bundle, artifact, root)
+
+    def test_rekor_v2_mutates_authenticated_checkpoint_not_unsigned_hints(self):
+        bundle = {"verificationMaterial": {"tlogEntries": [{"kindVersion": {"kind": "hashedrekord", "version": "0.0.2"}}]}}
+        names = {mutation.name for mutation in get_mutations_for_bundle(bundle)}
+        self.assertIn("checkpoint_wrong_root_hash", names)
+        self.assertNotIn("inclusion_proof_wrong_root_hash", names)
+        self.assertNotIn("integrated_time_future", names)
+        bundle["verificationMaterial"]["tlogEntries"][0]["kindVersion"]["version"] = "0.0.1"
+        names = {mutation.name for mutation in get_mutations_for_bundle(bundle)}
+        self.assertIn("inclusion_proof_wrong_root_hash", names)
+        self.assertIn("integrated_time_future", names)
+
+    def test_always_rejecting_executable_cannot_pass(self):
+        with tempfile.TemporaryDirectory() as directory:
+            script = Path(directory) / "reject.py"
+            script.write_text("print('Verification error: fake rejection')\nraise SystemExit(1)\n")
+            fuzzer = BundleMutationFuzzer(verifier_command=[sys.executable, str(script)])
+            with self.assertRaisesRegex(RuntimeError, "baseline"):
+                fuzzer.fuzz_bundle(Path("unused-bundle"), Path("unused-artifact"))
+
+
+if __name__ == "__main__":
+    unittest.main()