audit-core/scripts/mint-tenant-engine-sender.py

#!/usr/bin/env python3
"""Attended mint of the tenant-engine sender into the audit-core registry.

Silent: warden access --exec fails closed on any child stdout/stderr.
Never prints token values. Persist metadata evidence only.

  ./scripts/attended-mint-tenant-engine-sender.sh
"""
from __future__ import annotations

import base64
import json
import os
import secrets
import subprocess
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

REGISTRY = "platform/data/workloads/audit-core/senders"
SENDER = "tenant-engine"
TRADE = (
    "outbox drains after commit, not inside the mutation transaction; "
    "declared at tenant-engine/docs/evidence-emission.md"
)
EVIDENCE = Path("/home/worsch/audit-core/docs/evidence/2026-09-15-tenant-engine-sender-mint.json")
KUBECONFIG = os.environ.get("RAILIANCE01_KUBECONFIG", "/home/worsch/.kube/config-railiance01")
RECEIVER_IMAGE = (
    "forgejo.coulomb.social/coulomb/audit-core@"
    "sha256:ec15f63d49226bfe507af2bc38ffbd5e83f549ba2c6d1ae7a338e7e053f34615"
)


def _now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def _evidence(step: str, extra: dict | None = None) -> None:
    doc = {
        "step": step,
        "observed_at": _now(),
        "sender": SENDER,
        "registry": "platform/workloads/audit-core/senders",
        "credential_values_emitted": False,
    }
    if extra:
        doc.update(extra)
    EVIDENCE.parent.mkdir(parents=True, exist_ok=True)
    EVIDENCE.write_text(json.dumps(doc, indent=2) + "\n")


def _fail(reason: str, extra: dict | None = None) -> int:
    payload = {"reason": reason}
    if extra:
        payload.update(extra)
    _evidence("failed", payload)
    return 1


def bao(args: list[str], *, payload: dict | None = None) -> dict:
    env = os.environ.copy()
    env.pop("BAO_TOKEN", None)
    env.pop("VAULT_TOKEN", None)
    env.pop("OPENBAO_TOKEN", None)
    proc = subprocess.run(
        ["bao", *args],
        input=None if payload is None else json.dumps(payload).encode(),
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env=env,
        check=False,
    )
    if proc.returncode != 0:
        raise RuntimeError("bao_failed")
    return json.loads(proc.stdout.decode() or "{}")


def kube(args: list[str]) -> subprocess.CompletedProcess:
    return subprocess.run(
        ["kubectl", "--kubeconfig", KUBECONFIG, "--request-timeout=30s", *args],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
    )


def kube_json(args: list[str]) -> dict:
    proc = kube(args)
    if proc.returncode != 0:
        raise RuntimeError("kubectl_failed")
    return json.loads(proc.stdout.decode())


def sender_names(rows: list) -> list[str]:
    return [row.get("name") for row in rows if isinstance(row, dict) and row.get("name")]


def secret_sender_names() -> list[str]:
    secret = kube_json(["-n", "audit-core", "get", "secret", "audit-core-senders", "-o", "json"])
    raw = (secret.get("data") or {}).get("senders.json")
    if not raw:
        return []
    doc = json.loads(base64.b64decode(raw))
    if not isinstance(doc, list):
        return []
    return sender_names(doc)


def desired(token: str) -> dict:
    return {
        "name": SENDER,
        "tokens": [token],
        "sources": [SENDER],
        "tenants": ["*"],
        "may_write": True,
        "may_read": False,
        "secret_policy": "redact",
        "evidence_kind": "attributive",
        "completeness_trade": TRADE,
    }


def wait_secret_has_sender(timeout: int = 90) -> bool:
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            if SENDER in secret_sender_names():
                return True
        except Exception:
            pass
        time.sleep(3)
    return False


def wait_receiver_ready(timeout: int = 180) -> bool:
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            dep = kube_json(["-n", "audit-core", "get", "deploy", "audit-core", "-o", "json"])
            ready = dep.get("status", {}).get("readyReplicas")
            image = dep["spec"]["template"]["spec"]["containers"][0]["image"]
            if ready == 1 and image == RECEIVER_IMAGE:
                pods = kube_json(
                    [
                        "-n",
                        "audit-core",
                        "get",
                        "pods",
                        "-l",
                        "app.kubernetes.io/name=audit-core,app.kubernetes.io/component=receiver",
                        "-o",
                        "json",
                    ]
                )["items"]
                if (
                    len(pods) == 1
                    and not pods[0]["metadata"].get("deletionTimestamp")
                    and all(cs.get("ready") for cs in pods[0].get("status", {}).get("containerStatuses") or [])
                ):
                    return True
        except Exception:
            pass
        time.sleep(4)
    return False


def main() -> int:
    _evidence("started")
    home = Path.home()
    if home.parent.name != ".warden-attended-login":
        return _fail("attended_warden_envelope_required")
    if os.environ.get("BAO_TOKEN") or os.environ.get("VAULT_TOKEN"):
        return _fail("token_env_must_be_unset")
    try:
        identity = bao(["token", "lookup", "-format=json"])["data"]
    except Exception:
        return _fail("token_lookup_failed")
    policies = identity.get("policies") or []
    if "platform-admin" not in policies or "root" in policies:
        return _fail("attended_platform_admin_required")
    if kube(["get", "ns", "audit-core", "-o", "name"]).returncode != 0:
        return _fail("not_railiance01")
    if kube(["get", "ns", "core-hub-staging", "-o", "name"]).returncode == 0:
        return _fail("coulombcore_kubeconfig")

    try:
        raw = bao(["read", "-format=json", REGISTRY])["data"]
        version = raw["metadata"]["version"]
        body = raw["data"]
        rows = json.loads(body["senders.json"])
    except Exception:
        return _fail("registry_read_failed")
    if not isinstance(rows, list) or not rows:
        return _fail("registry_must_preserve_existing_senders")
    names = sender_names(rows)
    before = {"registry_version": version, "senders": names}
    _evidence("registry_read", before)

    existing = next((row for row in rows if row.get("name") == SENDER), None)
    minted = False
    if existing is not None:
        tokens = existing.get("tokens") or []
        if not tokens:
            return _fail("existing_tenant_engine_has_no_tokens")
        _evidence("already_present", {"registry_version": version, "senders": names})
    else:
        used = set()
        for row in rows:
            for token in row.get("tokens") or []:
                used.add(token)
        token = secrets.token_urlsafe(48)
        if token in used:
            return _fail("sender_token_collision")
        updated = list(rows)
        updated.append(desired(token))
        payload = dict(body, **{"senders.json": json.dumps(updated, separators=(",", ":"))})
        try:
            result = bao(
                ["write", "-format=json", REGISTRY, "-"],
                payload={"options": {"cas": version}, "data": payload},
            )
        except Exception:
            return _fail("registry_cas_failed")
        minted = True
        try:
            after = bao(["read", "-format=json", REGISTRY])["data"]
            after_rows = json.loads(after["data"]["senders.json"])
            after_names = sender_names(after_rows)
            after_version = after["metadata"]["version"]
        except Exception:
            return _fail("registry_readback_failed")
        if SENDER not in after_names or set(names) - {SENDER} != set(after_names) - {SENDER}:
            return _fail("registry_readback_mismatch", {"senders": after_names})
        names = after_names
        version = after_version
        _evidence(
            "registry_written",
            {
                "registry_version": version,
                "senders": names,
                "write_request_id": result.get("request_id"),
                "minted": True,
            },
        )

    if kube(
        [
            "-n",
            "audit-core",
            "annotate",
            "externalsecret",
            "audit-core-senders",
            f"force-sync={int(time.time())}",
            "--overwrite",
        ]
    ).returncode != 0:
        return _fail("externalsecret_annotate_failed")
    if not wait_secret_has_sender():
        return _fail("secret_sync_timeout", {"senders": names})
    _evidence("secret_synced", {"senders": secret_sender_names(), "minted": minted})

    if kube(["-n", "audit-core", "rollout", "restart", "deploy/audit-core"]).returncode != 0:
        return _fail("receiver_restart_failed")
    if not wait_receiver_ready():
        return _fail("receiver_not_ready_after_restart")
    live = secret_sender_names()
    if SENDER not in live:
        return _fail("secret_lost_tenant_engine", {"senders": live})
    _evidence(
        "done",
        {
            "senders": live,
            "has_tenant_engine_tokens": True,
            "receiver_ready": True,
            "minted": minted,
            "registry_version": version,
        },
    )
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except SystemExit:
        raise
    except Exception:
        _fail("contained_operation_failed")
        raise SystemExit(1)