Recover model provenance from verified backup receipt

2026-08-24 03:22:45 +02:00
parent 36a3c84699
commit 3627a05bfe
2 changed files with 140 additions and 0 deletions
@@ -163,3 +163,55 @@ def test_model_registry_status_migration_is_narrow_and_restores_write_guard() ->
    assert migration.count("DISABLE TRIGGER trg_source_registry_write_guard") == 1
    assert migration.count("ENABLE TRIGGER trg_source_registry_write_guard") == 1
    assert 'down_revision = "202608010001"' in migration
+
+
+def test_recovery_receipt_requires_hashed_backup_inventory(tmp_path: Path) -> None:
+    args = _args(tmp_path)
+    evidence = module.inspect_evidence(args)
+    receipt = tmp_path / "dry-run.json"
+    receipt.write_text(
+        json.dumps(
+            {
+                "status": "ready_to_apply",
+                "claim_boundary": module.CLAIM_BOUNDARY,
+                "evidence": evidence,
+            }
+        ),
+        encoding="utf-8",
+    )
+    storage_manifest = tmp_path / "storage-manifest.tsv"
+    missing_names = ("checkpoint", "training_summary", "training_args", "training_results")
+    rows = ["relative_path\tsize_bytes\tmtime_ns\tsha256"]
+    for name in missing_names:
+        original = Path(evidence["paths"][name])
+        evidence["paths"][name] = f"/app/storage/training/{original.name}"
+        rows.append(f"training/{original.name}\t1\t0\t{evidence['checksums'][name]}")
+    receipt.write_text(
+        json.dumps(
+            {
+                "status": "ready_to_apply",
+                "claim_boundary": module.CLAIM_BOUNDARY,
+                "evidence": evidence,
+            }
+        ),
+        encoding="utf-8",
+    )
+    storage_manifest.write_text("\n".join(rows) + "\n", encoding="utf-8")
+    backup_checksums = tmp_path / "CHECKSUMS.sha256"
+    backup_checksums.write_text(
+        f"{_sha(storage_manifest.read_bytes())}  storage-manifest.tsv\n",
+        encoding="utf-8",
+    )
+    args.evidence_receipt_path = str(receipt)
+    args.backup_storage_manifest_path = str(storage_manifest)
+    args.backup_checksums_path = str(backup_checksums)
+
+    recovered = module.inspect_evidence(args)
+
+    assert recovered["checksums"] == evidence["checksums"]
+    assert recovered["recovery_receipt"]["missing_artifacts_not_recreated"] == list(missing_names)
+
+    storage_manifest.write_text(storage_manifest.read_text(encoding="utf-8") + "tampered\n", encoding="utf-8")
+    exit_code, payload = module.migrate(args)
+    assert exit_code == 2
+    assert "backup storage manifest SHA-256 mismatch" in payload["message"]