#!/usr/bin/env python3 """Dry-run-first quarantine for old unreferenced derived/cache artifacts.""" from __future__ import annotations import argparse import hashlib import json import os from datetime import datetime, timezone from pathlib import Path from uuid import uuid4 from audit_data_operations import build_report from release_backup_guard import require_confirmation, verify_current_backup from app.core.config import get_settings from app.db.session import SessionLocal CONFIRMATION = "QUARANTINE_STORAGE_ARTIFACTS" def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def write_manifest(path: Path, payload: dict[str, object]) -> None: temporary = path.with_suffix(".json.partial") temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") temporary.replace(path) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--storage-root", type=Path) parser.add_argument("--minimum-age-days", type=int, default=7) parser.add_argument("--max-delete", type=int, default=25) parser.add_argument("--apply", action="store_true") parser.add_argument("--confirm") parser.add_argument("--backup-dir", type=Path) parser.add_argument("--backup-max-age-hours", type=float, default=24.0) parser.add_argument( "--quarantine-root", type=Path, help="Protected destination below the storage root (default: operator-evidence/cleanup-quarantine)", ) return parser.parse_args() def main() -> int: args = parse_args() if args.max_delete < 0: raise SystemExit("--max-delete must be greater than or equal to zero") storage_root = (args.storage_root or Path(get_settings().storage_root)).resolve() with SessionLocal() as db: report, candidates = build_report( storage_root, db, minimum_age_days=args.minimum_age_days, max_candidate_records=max(args.max_delete, 500), ) blocked_reason = None backup = None quarantined: list[dict[str, object]] = [] quarantine_manifest: Path | None = None if args.apply: require_confirmation(args.confirm, CONFIRMATION) if args.backup_dir is None: raise RuntimeError("--backup-dir is required with --apply") backup = verify_current_backup( args.backup_dir, max_age_hours=args.backup_max_age_hours, ) if len(candidates) > args.max_delete: blocked_reason = ( f"candidate_count {len(candidates)} exceeds --max-delete {args.max_delete}; " "review the dry run and raise the explicit limit" ) else: quarantine_root = ( args.quarantine_root or storage_root / "operator-evidence" / "cleanup-quarantine" ).resolve() protected_quarantine_root = ( storage_root / "operator-evidence" / "cleanup-quarantine" ).resolve() try: quarantine_root.relative_to(protected_quarantine_root) except ValueError as exc: raise RuntimeError( "--quarantine-root must remain below " "operator-evidence/cleanup-quarantine in --storage-root" ) from exc operation_id = f"cleanup-{datetime.now(timezone.utc):%Y%m%dT%H%M%SZ}-{uuid4().hex[:12]}" operation_root = quarantine_root / operation_id operation_root.mkdir(parents=True, exist_ok=False) quarantine_manifest = operation_root / "manifest.json" entries: list[dict[str, object]] = [] for candidate in candidates: if candidate.path.is_symlink(): raise RuntimeError(f"Cleanup candidate became a symlink: {candidate.relative_path}") try: candidate.path.resolve().relative_to(storage_root) except ValueError as exc: raise RuntimeError( f"Cleanup candidate escaped storage: {candidate.relative_path}" ) from exc destination = operation_root / "files" / candidate.relative_path current_size = candidate.path.stat().st_size if current_size != candidate.size_bytes: raise RuntimeError(f"Cleanup candidate changed size: {candidate.relative_path}") entries.append( { "relative_path": candidate.relative_path, "size_bytes": current_size, "sha256": sha256(candidate.path), "status": "planned", "quarantine_relative_path": destination.relative_to(storage_root).as_posix(), } ) manifest: dict[str, object] = { "schema_version": 1, "operation_id": operation_id, "created_at": datetime.now(timezone.utc).isoformat(), "state": "in_progress", "storage_root": str(storage_root), "backup_release_id": backup.release_id, "entries": entries, } write_manifest(quarantine_manifest, manifest) for candidate, entry in zip(candidates, entries, strict=True): destination = storage_root / str(entry["quarantine_relative_path"]) destination.parent.mkdir(parents=True, exist_ok=True) try: os.link(candidate.path, destination, follow_symlinks=False) except FileExistsError as exc: raise RuntimeError(f"Quarantine destination already exists: {destination}") from exc if ( not destination.is_file() or destination.stat().st_size != entry["size_bytes"] or sha256(destination) != entry["sha256"] ): destination.unlink(missing_ok=True) raise RuntimeError(f"Quarantine link verification failed: {candidate.relative_path}") entry["status"] = "linked" write_manifest(quarantine_manifest, manifest) candidate.path.unlink() entry["status"] = "quarantined" quarantined.append(dict(entry)) write_manifest(quarantine_manifest, manifest) manifest["state"] = "complete" manifest["completed_at"] = datetime.now(timezone.utc).isoformat() write_manifest(quarantine_manifest, manifest) payload = { "schema_version": 1, "mode": "apply" if args.apply else "dry-run", "storage_root": str(storage_root), "minimum_age_days": args.minimum_age_days, "max_delete": args.max_delete, "candidate_count": len(candidates), "candidate_bytes": sum(item.size_bytes for item in candidates), "candidates": [item.relative_path for item in candidates], "deleted_count": 0, "deleted": [], "quarantined_count": len(quarantined), "quarantined": quarantined, "quarantine_manifest": str(quarantine_manifest) if quarantine_manifest else None, "blocked_reason": blocked_reason, "protected_prefixes": report["cleanup"]["protected_prefixes"], "backup": ( { "release_id": backup.release_id, "created_at": backup.created_at.isoformat(), "age_hours": round(backup.age_hours, 3), "backup_tool_revision": backup.backup_tool_revision, } if backup else None ), } print(json.dumps(payload, indent=2, sort_keys=True)) return 1 if blocked_reason else 0 if __name__ == "__main__": raise SystemExit(main())