railiance-platform/scripts/mas_evidence_age_escrow.py

#!/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())