diff --git a/backend/tests/test_runtime_model_provenance_migration.py b/backend/tests/test_runtime_model_provenance_migration.py index cc970650..c13fd12d 100644 --- a/backend/tests/test_runtime_model_provenance_migration.py +++ b/backend/tests/test_runtime_model_provenance_migration.py @@ -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"] diff --git a/scripts/migrate_runtime_model_provenance.py b/scripts/migrate_runtime_model_provenance.py index a4571acf..ddebc3e4 100644 --- a/scripts/migrate_runtime_model_provenance.py +++ b/scripts/migrate_runtime_model_provenance.py @@ -63,6 +63,8 @@ def _require_equal(observed: Any, expected: Any, field: str) -> None: def inspect_evidence(args: argparse.Namespace) -> dict[str, Any]: + if args.evidence_receipt_path: + return inspect_recovery_receipt(args) paths = { "model": _required_file(args.model_path, "--model-path"), "checkpoint": _required_file(args.checkpoint_path, "--checkpoint-path"), @@ -112,6 +114,89 @@ def inspect_evidence(args: argparse.Namespace) -> dict[str, Any]: return evidence +def _checksum_manifest_entries(path: Path) -> dict[str, str]: + entries: dict[str, str] = {} + for line in path.read_text(encoding="utf-8").splitlines(): + checksum, separator, name = line.partition(" ") + if separator and len(checksum) == 64: + entries[name.strip()] = checksum.lower() + return entries + + +def _storage_manifest_entries(path: Path) -> dict[str, tuple[int, str]]: + entries: dict[str, tuple[int, str]] = {} + lines = path.read_text(encoding="utf-8").splitlines() + if not lines or lines[0] != "relative_path\tsize_bytes\tmtime_ns\tsha256": + raise ValueError("backup storage manifest has an unexpected header") + for line in lines[1:]: + fields = line.split("\t") + if len(fields) == 4: + entries[fields[0]] = (int(fields[1]), fields[3].lower()) + return entries + + +def inspect_recovery_receipt(args: argparse.Namespace) -> dict[str, Any]: + receipt_path = _required_file(args.evidence_receipt_path, "--evidence-receipt-path") + storage_manifest_path = _required_file( + args.backup_storage_manifest_path, "--backup-storage-manifest-path" + ) + backup_checksums_path = _required_file(args.backup_checksums_path, "--backup-checksums-path") + receipt = _json_object(receipt_path, "evidence receipt") + _require_equal(receipt.get("status"), "ready_to_apply", "evidence receipt status") + _require_equal(receipt.get("claim_boundary"), CLAIM_BOUNDARY, "evidence receipt claim boundary") + evidence = receipt.get("evidence") + if not isinstance(evidence, dict): + raise ValueError("evidence receipt does not contain an evidence object") + paths = evidence.get("paths") + checksums = evidence.get("checksums") + if not isinstance(paths, dict) or not isinstance(checksums, dict): + raise ValueError("evidence receipt paths/checksums are incomplete") + + backup_checksums = _checksum_manifest_entries(backup_checksums_path) + _require_equal( + backup_checksums.get(storage_manifest_path.name), + _sha256_file(storage_manifest_path), + "backup storage manifest SHA-256", + ) + storage_entries = _storage_manifest_entries(storage_manifest_path) + for name in ("checkpoint", "training_summary", "training_args", "training_results"): + original_path = Path(str(paths.get(name, ""))) + try: + relative_path = original_path.relative_to("/app/storage").as_posix() + except ValueError as exc: + raise ValueError(f"receipt {name} path is outside /app/storage: {original_path}") from exc + entry = storage_entries.get(relative_path) + if entry is None: + raise ValueError(f"backup storage manifest does not inventory receipt path: {relative_path}") + _require_equal(entry[1], checksums.get(name), f"backup {name} SHA-256") + + live_paths = { + "model": _required_file(args.model_path, "--model-path"), + "base_model": _required_file(args.base_model_path, "--base-model-path"), + "dataset_summary": _required_file(args.dataset_summary_path, "--dataset-summary-path"), + "dataset_yaml": _required_file(args.dataset_yaml_path, "--dataset-yaml-path"), + } + for name, live_path in live_paths.items(): + _require_equal(_sha256_file(live_path), checksums.get(name), f"live {name} SHA-256") + paths[name] = str(live_path) + _require_equal(checksums.get("checkpoint"), checksums.get("model"), "receipt checkpoint/model SHA-256") + evidence["recovery_receipt"] = { + "path": str(receipt_path), + "sha256": _sha256_file(receipt_path), + "backup_storage_manifest_path": str(storage_manifest_path), + "backup_storage_manifest_sha256": _sha256_file(storage_manifest_path), + "backup_checksums_path": str(backup_checksums_path), + "backup_checksums_sha256": _sha256_file(backup_checksums_path), + "missing_artifacts_not_recreated": [ + "checkpoint", + "training_summary", + "training_args", + "training_results", + ], + } + return evidence + + def _snapshot_key(*, model_id: str, model_sha256: str) -> str: return f"runtime-model-{model_id}-{model_sha256[:24]}" @@ -358,6 +443,9 @@ def parse_args(argv: list[str]) -> argparse.Namespace: parser.add_argument("--task-type", default="object_detection") parser.add_argument("--source-version", required=True) parser.add_argument("--framework-version", required=True) + parser.add_argument("--evidence-receipt-path") + parser.add_argument("--backup-storage-manifest-path") + parser.add_argument("--backup-checksums-path") parser.add_argument("--apply", action="store_true") parser.add_argument("--json", action="store_true") return parser.parse_args(argv)