127 lines
4.6 KiB
Python
127 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Verification guard shared by destructive GeoIntel operator commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class VerifiedBackup:
|
|
backup_dir: Path
|
|
release_id: str
|
|
created_at: datetime
|
|
age_hours: float
|
|
git_commit: str
|
|
|
|
|
|
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 _created_at(value: object) -> datetime:
|
|
if not isinstance(value, str):
|
|
raise RuntimeError("Backup manifest does not contain created_at")
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=timezone.utc)
|
|
return parsed.astimezone(timezone.utc)
|
|
|
|
|
|
def verify_current_backup(
|
|
backup_dir: str | Path,
|
|
*,
|
|
max_age_hours: float = 24.0,
|
|
now: datetime | None = None,
|
|
) -> VerifiedBackup:
|
|
"""Verify checksums and release metadata without mutating the backup."""
|
|
if max_age_hours <= 0:
|
|
raise ValueError("max_age_hours must be greater than zero")
|
|
|
|
root = Path(backup_dir).expanduser().resolve()
|
|
if not root.is_dir():
|
|
raise RuntimeError(f"Backup directory does not exist: {root}")
|
|
|
|
required = {
|
|
"manifest.json",
|
|
"CHECKSUMS.sha256",
|
|
"database.dump",
|
|
"database.list",
|
|
"database-metadata.tsv",
|
|
"table-counts.tsv",
|
|
"storage-manifest.tsv",
|
|
}
|
|
missing = sorted(name for name in required if not (root / name).is_file())
|
|
if missing:
|
|
raise RuntimeError(f"Backup is incomplete; missing: {', '.join(missing)}")
|
|
|
|
checksum_lines = (root / "CHECKSUMS.sha256").read_text(encoding="utf-8").splitlines()
|
|
checked: set[str] = set()
|
|
for line in checksum_lines:
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
expected, name = line.split(maxsplit=1)
|
|
except ValueError as exc:
|
|
raise RuntimeError("Backup checksum file has an invalid line") from exc
|
|
name = name.lstrip("*")
|
|
if "/" in name or "\\" in name or name in {".", ".."}:
|
|
raise RuntimeError(f"Backup checksum contains an unsafe path: {name}")
|
|
target = root / name
|
|
if not target.is_file():
|
|
raise RuntimeError(f"Backup checksum target is missing: {name}")
|
|
if _sha256(target) != expected.lower():
|
|
raise RuntimeError(f"Backup checksum mismatch: {name}")
|
|
checked.add(name)
|
|
|
|
unchecked = sorted((required - {"CHECKSUMS.sha256"}) - checked)
|
|
if unchecked:
|
|
raise RuntimeError(f"Backup checksum coverage is incomplete: {', '.join(unchecked)}")
|
|
|
|
manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
|
|
if manifest.get("schema_version") != 1 or manifest.get("read_only_source") is not True:
|
|
raise RuntimeError("Backup manifest schema or read-only marker is invalid")
|
|
if manifest.get("database_password_secure") is not True:
|
|
raise RuntimeError("Backup was made from an insecure database configuration")
|
|
if manifest.get("inventory_mode") != "sha256" or manifest.get("storage_inventory_requested") is not True:
|
|
raise RuntimeError("Destructive maintenance requires a SHA-256 storage inventory backup")
|
|
|
|
created = _created_at(manifest.get("created_at"))
|
|
current = now or datetime.now(timezone.utc)
|
|
if current.tzinfo is None:
|
|
current = current.replace(tzinfo=timezone.utc)
|
|
age_hours = (current.astimezone(timezone.utc) - created).total_seconds() / 3600
|
|
if age_hours < -0.1:
|
|
raise RuntimeError("Backup timestamp is in the future")
|
|
if age_hours > max_age_hours:
|
|
raise RuntimeError(
|
|
f"Backup is {age_hours:.1f} hours old; maximum allowed age is {max_age_hours:.1f} hours"
|
|
)
|
|
|
|
release_id = manifest.get("release_id")
|
|
git_commit = manifest.get("git_commit")
|
|
if not isinstance(release_id, str) or not release_id:
|
|
raise RuntimeError("Backup release id is missing")
|
|
if not isinstance(git_commit, str) or len(git_commit) < 7:
|
|
raise RuntimeError("Backup Git commit is missing")
|
|
return VerifiedBackup(
|
|
backup_dir=root,
|
|
release_id=release_id,
|
|
created_at=created,
|
|
age_hours=age_hours,
|
|
git_commit=git_commit,
|
|
)
|
|
|
|
|
|
def require_confirmation(actual: str | None, expected: str) -> None:
|
|
if actual != expected:
|
|
raise RuntimeError(f"Refusing destructive maintenance; pass --confirm {expected}")
|