diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c9f4b8d9..ee36f5fa 100644 --- a/.github/workflows/ci.yml +++ b/.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" diff --git a/crates/sigstore-verify/examples/verify_bundle.rs b/crates/sigstore-verify/examples/verify_bundle.rs index 069b0a2a..9f2ec53e 100644 --- a/crates/sigstore-verify/examples/verify_bundle.rs +++ b/crates/sigstore-verify/examples/verify_bundle.rs @@ -62,6 +62,8 @@ async fn main() { let mut identity_regexp: Option = None; let mut issuer: Option = None; let mut instance: Option = None; + let mut trusted_root_path: Option = None; + let mut tuf_root_path: Option = None; let mut staging = false; let mut positional: Vec = 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 Required certificate identity (exact match)"); eprintln!(" --certificate-identity-regexp Required certificate identity (regex)"); eprintln!(" --certificate-oidc-issuer Required OIDC issuer"); - eprintln!(" --instance Use a custom Sigstore instance"); + eprintln!(" --trusted-root Use a local Sigstore trusted root (offline)"); + eprintln!(" --instance --tuf-root Custom instance with trusted TUF bootstrap"); eprintln!(" --staging Use Sigstore staging instance"); eprintln!(" -h, --help Print this help message"); eprintln!(); diff --git a/tests/mutation_fuzzer.py b/tests/mutation_fuzzer.py index 476ff89e..c7db3962 100644 --- a/tests/mutation_fuzzer.py +++ b/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) diff --git a/tests/test_mutation_fuzzer.py b/tests/test_mutation_fuzzer.py new file mode 100644 index 00000000..34c1d589 --- /dev/null +++ b/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()