tegwick
08906da7d0
Some checks failed
Use-case regression gate / usecase-regressions (push) Has been cancelled
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
129 lines
4.4 KiB
Python
129 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Review or attend sealing of eight completed backup evidence metadata entries."""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
OBSERVATION = ROOT / "docs/release/backup-completed-evidence-custody-preflight-2026-10-31.json"
|
|
|
|
|
|
def prepare():
|
|
if not __debug__:
|
|
raise ValueError("optimized_python_refused")
|
|
observed = json.loads(OBSERVATION.read_text())
|
|
source = (ROOT / "scripts/backup_custody.py").read_text()
|
|
plan = dict(
|
|
schema="coulomb.completed-evidence-custody-proposal.v1",
|
|
targets=observed["targets"],
|
|
receipts=observed["receipts"],
|
|
runtime_sha256=hashlib.sha256(source.encode()).hexdigest(),
|
|
launcher_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
|
|
capacity_source_sha256=hashlib.sha256(
|
|
(ROOT / "scripts/backup_capacity.py").read_bytes()
|
|
).hexdigest(),
|
|
)
|
|
plan["proposal_digest"] = (
|
|
"sha256:"
|
|
+ hashlib.sha256(
|
|
json.dumps(plan, sort_keys=True, separators=(",", ":")).encode()
|
|
).hexdigest()
|
|
)
|
|
return plan, source
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
action = parser.add_mutually_exclusive_group()
|
|
action.add_argument("--preflight", action="store_true")
|
|
action.add_argument("--execute", action="store_true")
|
|
action.add_argument("--rollback", action="store_true")
|
|
parser.add_argument("--expect-digest")
|
|
args = parser.parse_args(argv)
|
|
plan, source = prepare()
|
|
digest = plan["proposal_digest"]
|
|
if args.expect_digest and args.expect_digest != digest:
|
|
raise SystemExit("Reviewed custody proposal changed; no action")
|
|
if not (args.preflight or args.execute or args.rollback):
|
|
print(
|
|
json.dumps(
|
|
dict(
|
|
status="prepared",
|
|
proposal_digest=digest,
|
|
entries=8,
|
|
permission_or_ownership_changed=False,
|
|
)
|
|
)
|
|
)
|
|
return 0
|
|
if args.execute or args.rollback:
|
|
if not args.expect_digest or not sys.stdin.isatty():
|
|
raise SystemExit("Founder terminal and exact digest required")
|
|
verb = "ROLLBACK" if args.rollback else "SEAL"
|
|
print(
|
|
"Restore the original metadata only."
|
|
if args.rollback
|
|
else "Seal eight completed evidence entries: root ownership, private modes; "
|
|
"bytes and paths preserved."
|
|
)
|
|
if input("Type " + verb + " followed by the full digest: ").strip() != verb + " " + digest:
|
|
raise SystemExit("Not confirmed; no action")
|
|
payload = dict(
|
|
plan=plan,
|
|
action="rollback" if args.rollback else "seal" if args.execute else "preflight",
|
|
capacity_source=(ROOT / "scripts/backup_capacity.py").read_text(),
|
|
)
|
|
bootstrap = (
|
|
source
|
|
+ """
|
|
import sys
|
|
p=json.load(sys.stdin)
|
|
try:
|
|
result=run(p['plan'],p['action'],p['capacity_source'])
|
|
except Exception as error:
|
|
result=dict(status='requires_inspection',proposal_digest=p['plan']['proposal_digest'],
|
|
failure_type=type(error).__name__,retry_automatically=False)
|
|
if isinstance(error,Refused):result['reason']=str(error)
|
|
print(json.dumps(result))
|
|
sys.exit(0 if result['status'] in ('completed_evidence_seal_preflight_verified',
|
|
'completed_evidence_private_root_custody_verified','original_metadata_restored_verified') else 1)
|
|
"""
|
|
)
|
|
try:
|
|
result = subprocess.run(
|
|
[
|
|
"ssh",
|
|
"-o",
|
|
"BatchMode=yes",
|
|
"-o",
|
|
"ConnectTimeout=10",
|
|
"railiance01",
|
|
"sudo -n python3 -c " + shlex.quote(bootstrap),
|
|
],
|
|
input=json.dumps(payload).encode(),
|
|
capture_output=True,
|
|
timeout=150,
|
|
)
|
|
value = json.loads(result.stdout)
|
|
except (ValueError, subprocess.TimeoutExpired):
|
|
print(
|
|
json.dumps(
|
|
dict(
|
|
status="remote_outcome_unconfirmed",
|
|
proposal_digest=digest,
|
|
retry_automatically=False,
|
|
)
|
|
)
|
|
)
|
|
return 1
|
|
print(json.dumps(value))
|
|
return int(result.returncode != 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|