Harden backup credentials and add durable image inventory publication

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883

scripts/refresh_live_images.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+"""Atomically accumulate non-secret image exports outside replaceable checkouts."""
+from __future__ import annotations
+
+import argparse
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import tempfile
+
+
+def read_images(path: Path) -> set[str]:
+    images = set()
+    for line in path.read_text(encoding="utf-8").splitlines():
+        image = line.strip()
+        if not image or image.startswith("#"):
+            continue
+        if any(c.isspace() for c in image) or "/" not in image:
+            raise ValueError("invalid image export")
+        images.add(image)
+    if not images:
+        raise ValueError("empty image export")
+    return images
+
+
+def refresh(output: Path, sources: list[Path]) -> dict:
+    # Read every required export before touching the last known good inventory.
+    images: set[str] = set()
+    for source in sources:
+        images.update(read_images(source))
+    if not images:
+        raise ValueError("at least one nonempty export is required")
+    output.parent.mkdir(parents=True, exist_ok=True)
+    with output.with_name(output.name + ".lock").open("a") as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX)
+        if output.exists():
+            images.update(read_images(output))
+        data = ("\n".join(sorted(images)) + "\n").encode()
+        temporary = None
+        try:
+            with tempfile.NamedTemporaryFile(dir=output.parent, delete=False) as stream:
+                temporary = Path(stream.name)
+                stream.write(data)
+                stream.flush()
+                os.fchmod(stream.fileno(), 0o644)
+                os.fsync(stream.fileno())
+            os.replace(temporary, output)
+            directory = os.open(output.parent, os.O_RDONLY | os.O_DIRECTORY)
+            try:
+                os.fsync(directory)
+            finally:
+                os.close(directory)
+        finally:
+            if temporary is not None:
+                temporary.unlink(missing_ok=True)
+    return {"path": str(output), "images": len(images), "sha256": hashlib.sha256(data).hexdigest()}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--source", type=Path, action="append", required=True)
+    args = parser.parse_args()
+    try:
+        receipt = refresh(args.output, args.source)
+    except (OSError, ValueError):
+        parser.exit(1, "inventory refresh failed; no successful publication receipt\n")
+    print(json.dumps(receipt, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())