194 lines
8.0 KiB
Python
194 lines
8.0 KiB
Python
#!/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())
|