da42f764c0
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883
76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
#!/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 re
|
|
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 not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:/@+-]*", 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())
|