101 lines
4.1 KiB
Python
101 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Create a non-secret capsule from a SQLite-to-Postgres restore proof."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
PRIVATE_FIELDS = ("source_sqlite_backup", "private_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 validate_restore_proof(proof: dict[str, Any]) -> list[str]:
|
|
problems = []
|
|
mismatches = proof.get("mismatched_tables") or []
|
|
if proof.get("artifact") != "sqlite_postgres_restore_canary":
|
|
problems.append(f"artifact={proof.get('artifact')}")
|
|
if proof.get("status") != "pass":
|
|
problems.append(f"status={proof.get('status')}")
|
|
if proof.get("source_integrity_check") != "ok":
|
|
problems.append(f"source_integrity_check={proof.get('source_integrity_check')}")
|
|
if mismatches:
|
|
problems.append(f"mismatched_tables={len(mismatches)}")
|
|
if proof.get("source_table_count") != proof.get("target_table_count"):
|
|
problems.append(
|
|
f"table_count source={proof.get('source_table_count')} target={proof.get('target_table_count')}"
|
|
)
|
|
if proof.get("source_total_rows") != proof.get("target_total_rows"):
|
|
problems.append(f"total_rows source={proof.get('source_total_rows')} target={proof.get('target_total_rows')}")
|
|
if not proof.get("generated_at_utc"):
|
|
problems.append("missing_generated_at_utc")
|
|
return problems
|
|
|
|
|
|
def build_capsule(proof_path: Path) -> dict[str, Any]:
|
|
proof = json.loads(proof_path.read_text())
|
|
problems = validate_restore_proof(proof)
|
|
capsule = {
|
|
"artifact": "sqlite_postgres_restore_canary_capsule",
|
|
"capsule_generated_at_utc": datetime.now(UTC).isoformat(),
|
|
"source_proof_generated_at_utc": proof.get("generated_at_utc"),
|
|
"source_proof_sha256": sha256(proof_path),
|
|
"status": "pass" if not problems else "fail",
|
|
"source_integrity_check": proof.get("source_integrity_check"),
|
|
"source_sqlite_backup_sha256": proof.get("source_sqlite_backup_sha256"),
|
|
"postgres_image": proof.get("postgres_image"),
|
|
"postgres_schema": proof.get("postgres_schema"),
|
|
"source_table_count": proof.get("source_table_count"),
|
|
"target_table_count": proof.get("target_table_count"),
|
|
"source_total_rows": proof.get("source_total_rows"),
|
|
"target_total_rows": proof.get("target_total_rows"),
|
|
"mismatched_table_count": len(proof.get("mismatched_tables") or []),
|
|
"conversion_notes": proof.get("conversion_notes", {}),
|
|
"conversion_stats": proof.get("conversion_stats", {}),
|
|
"redacted_fields": list(PRIVATE_FIELDS),
|
|
"problems": problems,
|
|
"not_proven_by_this_artifact": [
|
|
"private backup file availability",
|
|
"private generated SQL contents",
|
|
"GCP Cloud SQL import execution",
|
|
"GCP Cloud SQL target-count readback",
|
|
],
|
|
}
|
|
return capsule
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--proof", required=True, type=Path, help="Full sqlite-postgres restore canary proof JSON")
|
|
parser.add_argument("--output", required=True, type=Path, help="Redacted capsule JSON output path")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
capsule = build_capsule(args.proof)
|
|
rendered = json.dumps(capsule, indent=2, sort_keys=True) + "\n"
|
|
for private_field in PRIVATE_FIELDS:
|
|
if private_field in rendered and private_field not in json.dumps(capsule.get("redacted_fields", [])):
|
|
print(f"private field leaked into capsule: {private_field}", file=sys.stderr)
|
|
return 1
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(rendered)
|
|
print(rendered, end="")
|
|
return 0 if capsule["status"] == "pass" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|