tegwick
16c685f956
Some checks failed
Use-case regression gate / usecase-regressions (push) Has been cancelled
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
79 lines
3.4 KiB
Python
79 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Seal existing private date plan, current target pins and complete-store baseline.
|
|
|
|
Offline preparation only. It creates no approval and never contacts the cluster.
|
|
Acquire the target/baseline AFTER the separately supervised storage release.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from source_date_supervisor import Capsule, canonical, sha, sync_directory
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def prepare(plan, baseline, target, output):
|
|
output = output.absolute()
|
|
if (output.exists() or any(p.is_symlink() for p in [output, *output.parents])
|
|
or output.resolve().is_relative_to(ROOT)):
|
|
raise ValueError("Fresh private destination outside checkout required")
|
|
sources = {
|
|
"supervisor.py": ROOT / "scripts/source_date_supervisor.py",
|
|
"wire.py": ROOT / "scripts/source_date_wire.py",
|
|
**{name: ROOT / "coulomb_social/apps/spaces" / name for name in
|
|
("date_backfill.py", "date_backfill_journal.py", "date_backfill_operation.py")},
|
|
"plan.json": plan,
|
|
}
|
|
if any(p.is_symlink() or not p.is_file() or p.resolve().is_relative_to(output)
|
|
for p in [*sources.values(), baseline, target]):
|
|
raise ValueError("Regular existing inputs required")
|
|
content = {name: path.read_bytes() for name, path in sources.items()}
|
|
manifest = {"schema": "coulomb.supervised-date-capsule.v1",
|
|
"attempt_id": uuid.uuid4().hex,
|
|
"target": json.loads(target.read_bytes()),
|
|
"baseline": json.loads(baseline.read_bytes()),
|
|
"files": {name: sha(data) for name, data in content.items()},
|
|
"requires_founder_attended_execution": True,
|
|
"approval_status": "prepared_not_admitted"}
|
|
digest = "sha256:" + sha(canonical(manifest))
|
|
# A private temporary validation directory avoids publishing an invalid or
|
|
# partial capsule under the requested final name.
|
|
import tempfile
|
|
|
|
with tempfile.TemporaryDirectory(prefix="csoc-date-capsule-") as temporary:
|
|
stage = Path(temporary)
|
|
for name, data in {**content, "manifest.json": canonical(manifest)}.items():
|
|
(stage / name).write_bytes(data)
|
|
Capsule(stage, digest)
|
|
output.mkdir(mode=0o700)
|
|
for name, data in {**content, "manifest.json": canonical(manifest)}.items():
|
|
fd = os.open(output / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(fd, "wb") as stream:
|
|
stream.write(data)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
sync_directory(output)
|
|
sync_directory(output.parent)
|
|
return {"status": "prepared_not_admitted", "proposal_digest": digest,
|
|
"file_count": len(content) + 1, "production_access": False, "live_changes": False}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--plan", type=Path, required=True)
|
|
parser.add_argument("--baseline", type=Path, required=True)
|
|
parser.add_argument("--target", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
print(json.dumps(prepare(args.plan, args.baseline, args.target, args.output)))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception as exc:
|
|
raise SystemExit("Date capsule preparation refused: " + type(exc).__name__)
|