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

#!/usr/bin/env python3
"""Attended seed of the tenant-engine sender token into its dedicated KV path.

ops-mason MASON-WP-0007 / tenant-engine TEN-IN-0005, plan row 7. The registry
document holds every sender, and OpenBao policies scope by path, not by field,
so External Secrets reads a one-field copy instead:

  source:      platform/data/workloads/audit-core/senders   (sender "tenant-engine")
  destination: platform/data/workloads/tenant-engine/audit-core-sender  (field "token")

Read and write happen in this one process. The token is never printed, never
passed on a command line, never written to evidence; evidence holds versions,
booleans and counts only. Silent: warden access --exec fails closed on any
child stdout/stderr.

  ./scripts/attended-seed-tenant-engine-audit-core-sender.sh

Re-running is safe. Equal value: no write. Different value (rotation re-seed):
one check-and-set write against the version just read. No destination: cas 0.
When the registry holds several tokens for the sender (rotation overlap), the
last one is seeded, which is the newest appended.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path

REGISTRY = "platform/data/workloads/audit-core/senders"
SENDER = "tenant-engine"
DEST_DATA = "platform/data/workloads/tenant-engine/audit-core-sender"
DEST_META = "platform/metadata/workloads/tenant-engine/audit-core-sender"
FIELD = "token"
EVIDENCE_DIR = Path("/home/worsch/audit-core/docs/evidence")


class NotFound(Exception):
    """The path holds no value. Distinct from a failed call."""


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


def evidence_path() -> Path:
    return EVIDENCE_DIR / f"{_now()[:10]}-tenant-engine-audit-core-sender-seed.json"


def _evidence(step: str, extra: dict | None = None) -> None:
    doc = {
        "step": step,
        "observed_at": _now(),
        "sender": SENDER,
        "source": "platform/workloads/audit-core/senders",
        "destination": "platform/workloads/tenant-engine/audit-core-sender",
        "credential_values_emitted": False,
    }
    if extra:
        doc.update(extra)
    path = evidence_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    path.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()
    for name in ("BAO_TOKEN", "VAULT_TOKEN", "OPENBAO_TOKEN"):
        env.pop(name, 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:
        # stderr is inspected, never emitted.
        err = proc.stderr.decode(errors="replace").lower()
        if "no value found" in err or "404" in err:
            raise NotFound()
        raise RuntimeError("bao_failed")
    return json.loads(proc.stdout.decode() or "{}")


def pick_token(rows: list) -> str:
    """The sender's newest token. Raises ValueError with a reason, never a value."""
    row = next((r for r in rows if isinstance(r, dict) and r.get("name") == SENDER), None)
    if row is None:
        raise ValueError("sender_not_in_registry")
    tokens = [t for t in (row.get("tokens") or []) if isinstance(t, str) and t]
    if not tokens:
        raise ValueError("sender_has_no_tokens")
    return tokens[-1]


def seed(token: str, call=bao) -> dict:
    """Write token to the destination if needed. Returns metadata only."""
    try:
        current = call(["read", "-format=json", DEST_DATA])["data"]
        version = current["metadata"]["version"]
        existing = (current.get("data") or {}).get(FIELD)
    except NotFound:
        version, existing = 0, None
    if existing == token:
        return {"action": "already_seeded", "version": version}
    call(
        ["write", "-format=json", DEST_DATA, "-"],
        payload={"options": {"cas": version}, "data": {FIELD: token}},
    )
    after = call(["read", "-format=json", DEST_DATA])["data"]
    if (after.get("data") or {}).get(FIELD) != token:
        raise RuntimeError("readback_mismatch")
    return {
        "action": "created" if version == 0 else "reseeded",
        "version": after["metadata"]["version"],
    }


def main() -> int:
    _evidence("started")
    if Path.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")

    try:
        raw = bao(["read", "-format=json", REGISTRY])["data"]
        registry_version = raw["metadata"]["version"]
        rows = json.loads(raw["data"]["senders.json"])
    except Exception:
        return _fail("registry_read_failed")
    try:
        token = pick_token(rows)
    except ValueError as exc:
        return _fail(str(exc), {"registry_version": registry_version})
    _evidence("registry_read", {"registry_version": registry_version})

    try:
        result = seed(token)
    except Exception as exc:
        return _fail("seed_failed", {"error_class": exc.__class__.__name__})
    _evidence(
        "done",
        {
            "registry_version": registry_version,
            "destination_version": result["version"],
            "action": result["action"],
            "field": FIELD,
        },
    )
    return 0


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