railiance-apps/tools/vergabe_pilot_inventory_attended.py

#!/usr/bin/env python3
"""Founder-terminal review, signing and submission of the fixed pilot inventory.

Run on railiance01 from a synced railiance-apps checkout. Default only reviews
the packet. --submit uses the existing approver key and gated executor; the
request is staged as custodian-agt so the executor's submitter check is retained.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import stat
import subprocess
import uuid
from pathlib import Path

EXPECTED = "sha256:2014e62a84b0875fec155dbfd1be26963d9a39ab931092d26c5dfb9e627ffbd7"
PACKET = (Path(__file__).resolve().parents[1]
          / "docs/proposals/2026-11-05-vergabe-recovery-preflight/action.json")
GEXEC = Path("/usr/local/bin/gexec")
EXCHANGE = Path("/var/spool/gexec/exchange")
SPOOL = "/var/spool/gexec"

READ_RESULT = """
import json, pathlib, re, sys, time
request_id = sys.argv[1]
if not re.fullmatch(r'vergabe-inventory-[0-9a-f]{16}', request_id):
    raise SystemExit(1)
directory = pathlib.Path('/var/spool/gexec/results')
for attempt in range(60):
    matches = list(directory.glob(request_id + '.*.json'))
    if len(matches) == 1:
        result = json.loads(matches[0].read_text())
        print(json.dumps(result, indent=2))
        raise SystemExit(0 if result.get('status') == 'executed'
                         and result.get('exit_code') == 0 else 1)
    if len(matches) > 1:
        print(json.dumps({'status': 'ambiguous_result', 'request_id': request_id}))
        raise SystemExit(1)
    time.sleep(1)
print(json.dumps({'status': 'result_pending', 'request_id': request_id,
                  'instruction': 'Preserve this request id; inspect its result before resubmitting.'}))
raise SystemExit(2)
"""


def validated_packet(path=PACKET):
    packet = json.loads(path.read_text())
    encoded = json.dumps(packet["action"], sort_keys=True, separators=(",", ":"),
                         ensure_ascii=False, allow_nan=False).encode()
    digest = "sha256:" + hashlib.sha256(encoded).hexdigest()
    if digest != EXPECTED or packet.get("digest") != EXPECTED:
        raise ValueError("packet_digest_mismatch")
    return packet


def submit(packet, key=None, approval_file=None):
    info = GEXEC.resolve(strict=True).stat()
    if info.st_uid != 0 or not stat.S_ISREG(info.st_mode) or info.st_mode & 0o022:
        raise ValueError("admitted_root_owned_gexec_required")
    if EXCHANGE.is_symlink() or not EXCHANGE.is_dir():
        raise ValueError("admitted_exchange_required")
    request_id = "vergabe-inventory-" + uuid.uuid4().hex[:16]
    proposal = EXCHANGE / (request_id + ".action.json")
    approval = EXCHANGE / (request_id + ".approval.json")
    fd = os.open(proposal, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o644)
    with os.fdopen(fd, "w") as output:
        json.dump(packet, output, indent=2)
    if approval_file is not None:
        signed = json.loads(approval_file.read_text())
        if signed.get("payload", {}).get("digest") != EXPECTED:
            raise ValueError("approval_digest_mismatch")
        fd = os.open(approval, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o644)
        with os.fdopen(fd, "w") as output:
            json.dump(signed, output)
    else:
        # The existing CLI displays the exact action and asks the founder to sign.
        approved = subprocess.run([
            str(GEXEC), "approve", str(proposal), "--key", str(key),
            "--approver", "bernd", "--ttl", "900", "--out", str(approval),
        ], check=False)
        if approved.returncode:
            print(json.dumps({"status": "not_submitted", "request_id": request_id}))
            return 1
    queued = subprocess.run([
        "/usr/bin/sudo", "-u", "custodian-agt", str(GEXEC), "submit",
        "--spool", SPOOL, "--proposal", str(proposal), "--approval", str(approval),
        "--request-id", request_id,
    ], check=False)
    if queued.returncode:
        print(json.dumps({"status": "submission_unconfirmed", "request_id": request_id,
                          "instruction": "Inspect this request before creating another."}))
        return 1
    return subprocess.run([
        "/usr/bin/sudo", "-u", "custodian-agt", "/usr/bin/python3", "-c",
        READ_RESULT, request_id,
    ], check=False).returncode


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--submit", action="store_true")
    signing = parser.add_mutually_exclusive_group()
    signing.add_argument("--key", type=Path, help="Path to the existing gexec approver key")
    signing.add_argument("--approval", type=Path, help="Existing signed approval; keeps key in its custody")
    args = parser.parse_args(argv)
    packet = validated_packet()
    print(json.dumps({"status": "prepared", "digest": EXPECTED,
                      "scope": "Read-only Vergabe deployment/storage inventory",
                      "target": "kube:railiance01", "runtime_mutations": False}), flush=True)
    if not args.submit:
        return 0
    if not GEXEC.is_file():
        raise ValueError("admitted_gexec_missing")
    if args.key is None and args.approval is None:
        args.key = Path(input("Path to your existing gexec approver key: ").strip()).expanduser()
    return submit(packet, args.key, args.approval)


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except ValueError as error:
        failure = str(error)
        if failure not in {"packet_digest_mismatch", "approval_digest_mismatch", "admitted_gexec_missing",
                           "admitted_root_owned_gexec_required", "admitted_exchange_required"}:
            failure = "invalid_packet_or_runtime"
        print(json.dumps({"status": "launcher_failed", "failure": failure}))
        raise SystemExit(1)
    except Exception:
        print(json.dumps({"status": "launcher_failed", "failure": "attended_runtime_unavailable"}))
        raise SystemExit(1)