#!/usr/bin/env python3 """Create the MAS evidence age identity inside OpenBao custody. Run only inside the attended OpenBao platform-admin login: warden access openbao-platform-admin-login --exec -- \ python3 scripts/mas_evidence_age_escrow.py Generates one age X25519 identity in memory, writes it CAS=0 to platform/recovery/mas/age-identity (fields AGE_PRIVATE_KEY, AGE_RECIPIENT), reads it back, and proves it by an in-memory encrypt/decrypt round trip. The private key is never written to a file, argv, stdout, stderr or receipt. Refuses when the path already holds a value: a successor is a reviewed rotation, not a second run. Stdout and stderr stay silent; the receipt holds the public recipient, KV version and field names only. """ from __future__ import annotations import json import os import secrets import subprocess import sys import urllib.error import urllib.request from pathlib import Path ROOT = Path("/home/worsch/railiance-platform") RECEIPT_DIR = Path("/home/worsch/.local/state/railiance/receipts") RECEIPT_NAME = "mas-evidence-age-escrow.json" MOUNT = "platform" PATH = "recovery/mas/age-identity" FIELDS = ["AGE_PRIVATE_KEY", "AGE_RECIPIENT"] BAO_ADDR = "http://127.0.0.1:18200" class EscrowError(RuntimeError): def __init__(self, code, detail=None): super().__init__(code) self.code, self.detail = code, detail def _gate(): import importlib.util spec = importlib.util.spec_from_file_location( "mas_evidence_publisher_gate", ROOT / "scripts/mas_evidence_publisher_gate.py" ) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def generate(run=subprocess.run): """Return (private_key_line, recipient) without touching the filesystem.""" made = run(["age-keygen"], capture_output=True, timeout=30) if made.returncode: raise EscrowError("keygen_failed") lines = made.stdout.decode().splitlines() keys = [line.strip() for line in lines if line.startswith("AGE-SECRET-KEY-1")] if len(keys) != 1: raise EscrowError("keygen_output_rejected") recipient = derive(keys[0], run) return keys[0], recipient def derive(private_key, run=subprocess.run): out = run(["age-keygen", "-y"], input=(private_key + "\n").encode(), capture_output=True, timeout=30) lines = out.stdout.decode().split() if out.returncode or len(lines) != 1 or not lines[0].startswith("age1"): raise EscrowError("recipient_derivation_failed") return lines[0] def round_trip(private_key, recipient, run=subprocess.run): """Encrypt a random probe to the recipient and decrypt it with the key from stdin.""" probe = secrets.token_bytes(32) sealed = run(["age", "-r", recipient], input=probe, capture_output=True, timeout=30) if sealed.returncode: raise EscrowError("probe_encrypt_failed") read_fd, write_fd = os.pipe() try: os.write(write_fd, (private_key + "\n").encode()) os.close(write_fd) write_fd = None opened = run(["age", "-d", "-i", f"/dev/fd/{read_fd}"], input=sealed.stdout, capture_output=True, timeout=30, pass_fds=(read_fd,)) finally: os.close(read_fd) if write_fd is not None: os.close(write_fd) if opened.returncode or opened.stdout != probe: raise EscrowError("probe_round_trip_failed") return True class Bao: def __init__(self, token, addr=BAO_ADDR): self.token, self.addr = token, addr self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) def call(self, method, path, body=None): request = urllib.request.Request( self.addr + "/v1/" + path, data=None if body is None else json.dumps(body).encode(), method=method, headers={"X-Vault-Token": self.token, "Content-Type": "application/json"}, ) try: with self.opener.open(request, timeout=20) as response: raw = response.read(65537) return response.status, (json.loads(raw) if raw else {}) except urllib.error.HTTPError as exc: return exc.code, {} except (urllib.error.URLError, OSError): raise EscrowError("openbao_request_failed") from None def escrow(bao, generate_fn=generate, round_trip_fn=round_trip, derive_fn=derive): status, _ = bao.call("GET", f"{MOUNT}/metadata/{PATH}") if status == 200: raise EscrowError("existing_identity_requires_rotation_review") if status != 404: raise EscrowError("metadata_precheck_failed", {"http_status": status}) private_key, recipient = generate_fn() round_trip_fn(private_key, recipient) status, body = bao.call("POST", f"{MOUNT}/data/{PATH}", { "options": {"cas": 0}, "data": {"AGE_PRIVATE_KEY": private_key, "AGE_RECIPIENT": recipient}, }) version = (body.get("data") or {}).get("version") if isinstance(body, dict) else None if status != 200 or version != 1: raise EscrowError("escrow_write_failed", {"http_status": status}) status, body = bao.call("GET", f"{MOUNT}/data/{PATH}?version=1") data = ((body.get("data") or {}).get("data")) if isinstance(body, dict) else None if status != 200 or not isinstance(data, dict) or sorted(data) != FIELDS: raise EscrowError("escrow_readback_failed", {"http_status": status}) stored_key = data["AGE_PRIVATE_KEY"] matches = stored_key == private_key and data["AGE_RECIPIENT"] == recipient if not matches or derive_fn(stored_key) != recipient: raise EscrowError("escrow_readback_mismatch") round_trip_fn(stored_key, recipient) del private_key, stored_key, data return { "status": "escrowed", "path": f"{MOUNT}/{PATH}", "kv_version": 1, "cas": 0, "field_names": FIELDS, "recipient": recipient, "readback_matches": True, "round_trip_from_custody": True, "private_key_on_disk": False, "values_recorded": False, } def main(argv=None): args = list(sys.argv[1:] if argv is None else argv) sys.stdout = open(os.devnull, "w") sys.stderr = open(os.devnull, "w") try: if args: raise EscrowError("usage_refused") gate = _gate() addr = os.environ.get("BAO_ADDR") or os.environ.get("VAULT_ADDR") or BAO_ADDR if addr.rstrip("/") != BAO_ADDR: raise EscrowError("bao_addr_mismatch") token, source = gate.parent_token() if not token: raise EscrowError("parent_token_" + source) receipt = escrow(Bao(token)) except EscrowError as exc: receipt = {"status": "failed", "code": exc.code, "values_recorded": False} if exc.detail: receipt["detail"] = exc.detail except Exception as exc: receipt = {"status": "failed", "code": "contained_escrow_failure", "error_type": type(exc).__name__, "values_recorded": False} RECEIPT_DIR.mkdir(mode=0o700, exist_ok=True) path = RECEIPT_DIR / RECEIPT_NAME fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(fd, "w") as out: json.dump(receipt, out, indent=2, sort_keys=True) return 0 if receipt.get("status") == "escrowed" else 71 if __name__ == "__main__": sys.exit(main())