"""Prepare/read-only-check a sealed release; live mode is attended founder work. This launcher does not mint an approval or bypass gated-executor. The existing gated-executor profile covers Kubernetes, not the root-owned backup helper refresh. Only the founder runs --execute in their own terminal after reviewing the packet. """ from __future__ import annotations import argparse import hashlib import io import json import os import shlex import subprocess import sys import tarfile from pathlib import Path ROOT = Path(__file__).resolve().parents[1] PLAN = ROOT / "release/coulomb-identity-export-23ffe2c.json" def canonical(value): return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() def prepare(plan_path=None): plan_path = plan_path or PLAN if plan_path.is_symlink() or not plan_path.resolve().is_relative_to(ROOT.resolve()): raise RuntimeError("reviewed_plan_inside_owner_repository_required") plan = json.loads(plan_path.read_text()) values = plan.get("values_file", "release/identity-export-values.yaml") snapshot = plan.get("snapshot_file", "release/identity-export-snapshot.py") evidence = plan.get("evidence_directory", "release/identity-export-evidence") for name in (values, snapshot, evidence): if (not isinstance(name, str) or Path(name).is_absolute() or ".." in Path(name).parts or not name.startswith("release/")): raise RuntimeError("reviewed_release_paths_required") if not all((ROOT / name).is_file() for name in (values, snapshot)): raise RuntimeError("reviewed_release_artifacts_missing") paths = [ ROOT / name for name in ( "tools/coulomb-image-release.py", "tools/coulomb_image_release_node.py", "tools/coulomb_image_probe.py", values, snapshot, ) ] paths += sorted((ROOT / "charts/coulomb-social").rglob("*")) paths += sorted((ROOT / evidence).glob("*.json")) paths = [p for p in paths if p.is_file()] if any(p.is_symlink() or not p.resolve().is_relative_to(ROOT) for p in paths): raise RuntimeError("regular_reviewed_files_required") content = {str(p.relative_to(ROOT)): p.read_bytes() for p in paths} plan["files"] = { name: hashlib.sha256(data).hexdigest() for name, data in content.items() } for name, expected in plan["rehearsal_evidence_sha256"].items(): if plan["files"].get(evidence + "/" + name) != expected: raise RuntimeError("reviewed_candidate_evidence_changed") digest = "sha256:" + hashlib.sha256(canonical(plan)).hexdigest() content["manifest.json"] = canonical(plan) return plan, digest, content def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--preflight", action="store_true") parser.add_argument("--execute", action="store_true") parser.add_argument("--expect-digest") parser.add_argument("--output", type=Path) parser.add_argument("--plan", type=Path, default=PLAN) args = parser.parse_args() if args.preflight and args.execute: raise SystemExit("choose read-only preflight or attended execution") plan, digest, content = prepare(args.plan) if args.expect_digest and args.expect_digest != digest: raise SystemExit("reviewed proposal changed; do not execute") if not args.preflight and not args.execute: print( json.dumps( { "status": "prepared", "proposal_digest": digest, "file_count": len(content), "candidate": plan["candidate_image"], "live_changes": False, } ) ) return if args.execute: if not args.expect_digest or not sys.stdin.isatty(): raise SystemExit( "founder must review and run the exact digest in their own terminal" ) print("Reviewed proposal: " + digest) print( "Coulomb-only image update; existing database schema retained, fresh paired backup, postflight and backup image-pin refresh." ) if ( input("Type DEPLOY followed by the full digest to proceed: ").strip() != "DEPLOY " + digest ): raise SystemExit("not approved; no live change") buffer = io.BytesIO() with tarfile.open(fileobj=buffer, mode="w:") as archive: for name, data in content.items(): entry = tarfile.TarInfo(name) entry.size, entry.mode = len(data), 0o600 archive.addfile(entry, io.BytesIO(data)) remote = "/tmp/csoc-reviewed-image-" + digest.removeprefix("sha256:") receiver = r""" import hashlib,io,json,os,pathlib,sys,tarfile root=pathlib.Path(sys.argv[1]);expected=sys.argv[2] raw=sys.stdin.buffer.read(2*1024**2+1);assert len(raw)<=2*1024**2 with tarfile.open(fileobj=io.BytesIO(raw),mode='r:') as archive: entries=archive.getmembers();assert len({e.name for e in entries})==len(entries) assert all(e.isfile() and not pathlib.Path(e.name).is_absolute() and '..' not in pathlib.Path(e.name).parts for e in entries) data={e.name:archive.extractfile(e).read() for e in entries} plan=json.loads(data['manifest.json']) canonical=json.dumps(plan,sort_keys=True,separators=(',',':')).encode() assert 'sha256:'+hashlib.sha256(canonical).hexdigest()==expected assert set(data)==set(plan['files'])|{'manifest.json'} for name,digest in plan['files'].items():assert hashlib.sha256(data[name]).hexdigest()==digest root.mkdir(mode=0o700,exist_ok=True) assert root.lstat().st_uid==os.getuid() and root.lstat().st_mode&0o077==0 for name,value in data.items(): target=root/name;target.parent.mkdir(parents=True,exist_ok=True,mode=0o700) if target.exists():assert not target.is_symlink() and target.read_bytes()==value;continue fd=os.open(target,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600) with os.fdopen(fd,'wb') as stream:stream.write(value) """ staged = subprocess.run( [ "ssh", "railiance01", "python3 -c " + shlex.quote(receiver) + " " + shlex.quote(remote) + " " + shlex.quote(digest), ], input=buffer.getvalue(), capture_output=True, timeout=60, check=False, ) if staged.returncode: raise SystemExit("reviewed capsule staging refused; no live change") node_hash = plan["files"]["tools/coulomb_image_release_node.py"] guard = ( "import hashlib,pathlib,sys;p=pathlib.Path(" + repr(remote + "/tools/coulomb_image_release_node.py") + ");" "b=p.read_bytes();assert hashlib.sha256(b).hexdigest()==" + repr(node_hash) + ";" "sys.argv=['reviewed-release','--capsule'," + repr(remote) + ",'--expect-digest'," + repr(digest) + "," + repr("--execute" if args.execute else "--preflight") + "];exec(compile(b,str(p),'exec'),{'__name__':'__main__'})" ) result = subprocess.run( ["ssh", "railiance01", "sudo -n python3 -c " + shlex.quote(guard)], capture_output=True, timeout=3600, check=False, ) try: receipt = json.loads(result.stdout) except (ValueError, TypeError): receipt = { "status": "remote_execution_unconfirmed", "proposal_digest": digest, "journal": "/var/lib/coulomb-social-backup/releases/" + digest.removeprefix("sha256:") + "/receipt.json", } if args.output: fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "w") as stream: json.dump(receipt, stream, indent=2) print(json.dumps(receipt)) if result.returncode: raise SystemExit(1) if __name__ == "__main__": main()