up

bench/tools/verify.py

@@ -0,0 +1,333 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: AGPL-3.0-or-later
+# BENCH-AUTO-401-019: Offline VEX proof bundle verifier
+
+"""
+Offline verification of VEX proof bundles without network access.
+
+Validates:
+- DSSE envelope structure
+- Payload type and format
+- Evidence hash references
+- Justification catalog membership
+- CAS hash verification
+
+Usage:
+    python bench/tools/verify.py --bundle PATH [--cas-root PATH] [--catalog PATH]
+"""
+
+import argparse
+import base64
+import hashlib
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+
+class VerificationResult:
+    """Result of a verification check."""
+
+    def __init__(self, passed: bool, message: str, details: str = ""):
+        self.passed = passed
+        self.message = message
+        self.details = details
+
+    def __str__(self):
+        status = "\033[0;32m✓\033[0m" if self.passed else "\033[0;31m✗\033[0m"
+        result = f"{status} {self.message}"
+        if self.details:
+            result += f"\n    {self.details}"
+        return result
+
+
+def sha256_hex(data: bytes) -> str:
+    """Compute SHA-256 hash."""
+    return hashlib.sha256(data).hexdigest()
+
+
+def blake3_hex(data: bytes) -> str:
+    """Compute BLAKE3-256 hash (fallback to SHA-256)."""
+    try:
+        import blake3
+        return "blake3:" + blake3.blake3(data).hexdigest()
+    except ImportError:
+        return "sha256:" + sha256_hex(data)
+
+
+def load_json(path: Path) -> dict | None:
+    """Load JSON file."""
+    try:
+        with open(path, 'r', encoding='utf-8') as f:
+            return json.load(f)
+    except (json.JSONDecodeError, FileNotFoundError) as e:
+        return None
+
+
+def verify_dsse_structure(dsse: dict) -> list[VerificationResult]:
+    """Verify DSSE envelope structure."""
+    results = []
+
+    # Check required fields
+    if "payloadType" not in dsse:
+        results.append(VerificationResult(False, "Missing payloadType"))
+    else:
+        results.append(VerificationResult(True, f"payloadType: {dsse['payloadType']}"))
+
+    if "payload" not in dsse:
+        results.append(VerificationResult(False, "Missing payload"))
+    else:
+        results.append(VerificationResult(True, "payload present"))
+
+    if "signatures" not in dsse or not dsse["signatures"]:
+        results.append(VerificationResult(False, "Missing or empty signatures"))
+    else:
+        sig_count = len(dsse["signatures"])
+        results.append(VerificationResult(True, f"Found {sig_count} signature(s)"))
+
+        # Check for placeholder signatures
+        for i, sig in enumerate(dsse["signatures"]):
+            sig_value = sig.get("sig", "")
+            if sig_value.startswith("PLACEHOLDER"):
+                results.append(VerificationResult(
+                    False,
+                    f"Signature {i} is placeholder",
+                    "Bundle needs actual signing before deployment"
+                ))
+            else:
+                keyid = sig.get("keyid", "unknown")
+                results.append(VerificationResult(True, f"Signature {i} keyid: {keyid}"))
+
+    return results
+
+
+def decode_payload(dsse: dict) -> tuple[dict | None, list[VerificationResult]]:
+    """Decode DSSE payload."""
+    results = []
+
+    payload_b64 = dsse.get("payload", "")
+    if not payload_b64:
+        results.append(VerificationResult(False, "Empty payload"))
+        return None, results
+
+    try:
+        payload_bytes = base64.b64decode(payload_b64)
+        payload = json.loads(payload_bytes)
+        results.append(VerificationResult(True, "Payload decoded successfully"))
+        return payload, results
+    except Exception as e:
+        results.append(VerificationResult(False, f"Failed to decode payload: {e}"))
+        return None, results
+
+
+def verify_openvex(payload: dict) -> list[VerificationResult]:
+    """Verify OpenVEX document structure."""
+    results = []
+
+    # Check OpenVEX context
+    context = payload.get("@context", "")
+    if "openvex" in context.lower():
+        results.append(VerificationResult(True, f"OpenVEX context: {context}"))
+    else:
+        results.append(VerificationResult(False, f"Unexpected context: {context}"))
+
+    # Check statements
+    statements = payload.get("statements", [])
+    if not statements:
+        results.append(VerificationResult(False, "No VEX statements"))
+    else:
+        results.append(VerificationResult(True, f"Contains {len(statements)} statement(s)"))
+
+        for i, stmt in enumerate(statements):
+            vuln = stmt.get("vulnerability", {})
+            vuln_id = vuln.get("name", vuln.get("@id", "unknown"))
+            status = stmt.get("status", "unknown")
+            results.append(VerificationResult(
+                True,
+                f"Statement {i}: {vuln_id} -> {status}"
+            ))
+
+    return results
+
+
+def verify_evidence_hashes(payload: dict, cas_root: Path | None) -> list[VerificationResult]:
+    """Verify evidence hash references against CAS."""
+    results = []
+
+    statements = payload.get("statements", [])
+    for stmt in statements:
+        impact = stmt.get("impact_statement", "")
+        if "Evidence hash:" in impact:
+            hash_value = impact.split("Evidence hash:")[1].strip()
+            results.append(VerificationResult(True, f"Evidence hash: {hash_value[:16]}..."))
+
+            # Verify against CAS if root provided
+            if cas_root and cas_root.exists():
+                # Look for reachability.json in CAS
+                reach_file = cas_root / "reachability.json"
+                if reach_file.exists():
+                    with open(reach_file, 'rb') as f:
+                        content = f.read()
+                    actual_hash = blake3_hex(content)
+
+                    if actual_hash == hash_value or hash_value in actual_hash:
+                        results.append(VerificationResult(True, "Evidence hash matches CAS"))
+                    else:
+                        results.append(VerificationResult(
+                            False,
+                            "Evidence hash mismatch",
+                            f"Expected: {hash_value[:32]}..., Got: {actual_hash[:32]}..."
+                        ))
+
+    return results
+
+
+def verify_catalog_membership(payload: dict, catalog_path: Path) -> list[VerificationResult]:
+    """Verify justification is in catalog."""
+    results = []
+
+    if not catalog_path.exists():
+        results.append(VerificationResult(False, f"Catalog not found: {catalog_path}"))
+        return results
+
+    catalog = load_json(catalog_path)
+    if catalog is None:
+        results.append(VerificationResult(False, "Failed to load catalog"))
+        return results
+
+    # Extract catalog entries
+    entries = catalog if isinstance(catalog, list) else catalog.get("entries", [])
+    catalog_ids = {e.get("id", "") for e in entries}
+
+    # Check each statement's justification
+    statements = payload.get("statements", [])
+    for stmt in statements:
+        justification = stmt.get("justification")
+        if justification:
+            if justification in catalog_ids:
+                results.append(VerificationResult(
+                    True,
+                    f"Justification '{justification}' in catalog"
+                ))
+            else:
+                results.append(VerificationResult(
+                    False,
+                    f"Justification '{justification}' not in catalog"
+                ))
+
+    return results
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Offline VEX proof bundle verifier"
+    )
+    parser.add_argument(
+        "--bundle",
+        type=Path,
+        required=True,
+        help="Path to DSSE bundle file"
+    )
+    parser.add_argument(
+        "--cas-root",
+        type=Path,
+        default=None,
+        help="Path to CAS evidence directory"
+    )
+    parser.add_argument(
+        "--catalog",
+        type=Path,
+        default=Path("docs/benchmarks/vex-justifications.catalog.json"),
+        help="Path to justification catalog"
+    )
+
+    args = parser.parse_args()
+
+    # Resolve paths
+    repo_root = Path(__file__).parent.parent.parent
+    bundle_path = args.bundle if args.bundle.is_absolute() else repo_root / args.bundle
+    catalog_path = args.catalog if args.catalog.is_absolute() else repo_root / args.catalog
+    cas_root = args.cas_root if args.cas_root and args.cas_root.is_absolute() else (
+        repo_root / args.cas_root if args.cas_root else None
+    )
+
+    print(f"Verifying: {bundle_path}")
+    print("")
+
+    all_results = []
+    passed = 0
+    failed = 0
+
+    # Load DSSE bundle
+    dsse = load_json(bundle_path)
+    if dsse is None:
+        print("\033[0;31m✗\033[0m Failed to load bundle")
+        return 1
+
+    # Verify DSSE structure
+    print("DSSE Structure:")
+    results = verify_dsse_structure(dsse)
+    for r in results:
+        print(f"  {r}")
+        if r.passed:
+            passed += 1
+        else:
+            failed += 1
+    all_results.extend(results)
+
+    # Decode payload
+    print("\nPayload:")
+    payload, results = decode_payload(dsse)
+    for r in results:
+        print(f"  {r}")
+        if r.passed:
+            passed += 1
+        else:
+            failed += 1
+    all_results.extend(results)
+
+    if payload:
+        # Verify OpenVEX structure
+        payload_type = dsse.get("payloadType", "")
+        if "openvex" in payload_type.lower():
+            print("\nOpenVEX:")
+            results = verify_openvex(payload)
+            for r in results:
+                print(f"  {r}")
+                if r.passed:
+                    passed += 1
+                else:
+                    failed += 1
+            all_results.extend(results)
+
+        # Verify evidence hashes
+        print("\nEvidence:")
+        results = verify_evidence_hashes(payload, cas_root)
+        for r in results:
+            print(f"  {r}")
+            if r.passed:
+                passed += 1
+            else:
+                failed += 1
+        all_results.extend(results)
+
+        # Verify catalog membership
+        print("\nCatalog:")
+        results = verify_catalog_membership(payload, catalog_path)
+        for r in results:
+            print(f"  {r}")
+            if r.passed:
+                passed += 1
+            else:
+                failed += 1
+        all_results.extend(results)
+
+    # Summary
+    print(f"\n{'='*40}")
+    print(f"Passed: {passed}, Failed: {failed}")
+
+    return 0 if failed == 0 else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())