railiance-platform/scripts/workload_kv_write_next_version.py

#!/usr/bin/env python3
"""workload-kv-write-next-version CCRs: a workload writes the next version of one KV entry.

NetKingdom NK-WP-0056 (bounded executor, hub request 9d08288d). A non-human
executor rotates one existing value under a founder admission: it reads the
current version, writes the next version with CAS on that version, and may
revert by writing the prior value as a further version. It never creates a
path, reads metadata, deletes, destroys or lists.

The contract is source-only: validation, the exact policy HCL and an attended
procedure note. The policy, Kubernetes auth role and key-level ``cas_required``
are applied by the attended platform-admin from a reviewed commit, as with
attended-service-credential-delivery. Nothing here calls OpenBao or Kubernetes,
or reads, prints or accepts a secret value.
"""
from __future__ import annotations

import re
from typing import Any

REQUEST_TYPE = "workload-kv-write-next-version"
DOC = "docs/credential-lane-designs/bounded-executor.md"
APPROVED = {"approved", "applied", "verified", "active"}
KV_RE = re.compile(r"^platform/workloads/[a-z0-9][a-z0-9-]*(/[a-z0-9][a-z0-9-]*)+$")
NAME_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,62}$")
FIELD_RE = re.compile(r"^[A-Z][A-Z0-9_]*$")
OPENBAO_KEYS = {"mount", "kv_path", "fields", "policy_name", "policy_file", "write", "auth"}
WRITE = {"mode": "next-version", "cas_required": True, "create": False, "metadata_read": False,
         "delete": False, "token_self_lifecycle": True}
TTL = "15m"
AUTH_FIXED = {"method": "kubernetes", "mount": "kubernetes", "audience": "openbao", "ttl": TTL,
              "token_max_ttl": TTL, "token_explicit_max_ttl": TTL, "token_no_default_policy": True}
DISALLOWED_POLICIES = {"root", "default", "platform-admin"}


def policy_hcl(ccr: dict[str, Any]) -> str:
    """Exact policy: read and update of one data path; no create, metadata, delete or list.

    No required_parameters: OpenBao applies it to reads as well, so a GET (which carries no `options`) is
    refused and the executor cannot learn the version it must name (NK-WP-0056-T12 synthetic pilot,
    2026-10-31). The key's cas_required is what refuses a write without options.cas."""
    openbao = ccr["openbao"]
    suffix = openbao["kv_path"][len("platform/"):]
    return (f'path "platform/data/{suffix}" {{\n'
            '  capabilities = ["read", "update"]\n'
            '  allowed_parameters = {\n    "data" = []\n    "options" = []\n  }\n'
            "}\n\n"
            'path "auth/token/lookup-self" {\n  capabilities = ["read"]\n}\n\n'
            'path "auth/token/revoke-self" {\n  capabilities = ["update"]\n}\n')


def _approvals(ccr: dict[str, Any]) -> set[str]:
    comments = (ccr.get("review") or {}).get("comments") or []
    return {str(c.get("reviewer", "")).removeprefix("User (").removesuffix(")")
            for c in comments if c.get("decision") == "approved"}


def validate_contract(ccr: dict[str, Any]) -> list[str]:
    errors: list[str] = []
    if ccr.get("request_type") != REQUEST_TYPE:
        errors.append("unsupported write-next-version contract")
    openbao = ccr.get("openbao")
    if not isinstance(openbao, dict):
        return errors + ["openbao must be an object"]
    extra = set(openbao) - OPENBAO_KEYS
    if extra:
        errors.append("openbao declares unsupported keys: " + ", ".join(sorted(extra)))
    kv_path = str(openbao.get("kv_path", ""))
    if openbao.get("mount") != "platform" or not KV_RE.match(kv_path):
        errors.append("openbao.kv_path must be one exact platform/workloads/... path on mount platform")
    fields = openbao.get("fields")
    if not isinstance(fields, list) or not fields or not all(isinstance(f, str) and FIELD_RE.match(f) for f in fields):
        errors.append("openbao.fields must list upper-case field names")
    policy = openbao.get("policy_name")
    if not isinstance(policy, str) or not NAME_RE.match(policy) or policy in DISALLOWED_POLICIES:
        errors.append("openbao.policy_name must be a dedicated lower-case policy name")
    elif openbao.get("policy_file") != f"openbao/policies/{policy}.hcl":
        errors.append("openbao.policy_file must be openbao/policies/<policy_name>.hcl")
    if openbao.get("write") != WRITE:
        errors.append("openbao.write must be next-version with cas_required, no create, metadata or delete, "
                      "and token self-lifecycle")
    auth = openbao.get("auth")
    if not isinstance(auth, dict):
        errors.append("openbao.auth must be an object")
    else:
        for key, value in AUTH_FIXED.items():
            if auth.get(key) != value:
                errors.append(f"openbao.auth.{key} must be {value}")
        if not NAME_RE.match(str(auth.get("role", ""))):
            errors.append("openbao.auth.role must be a dedicated role name")
        if auth.get("policies") != [policy]:
            errors.append("openbao.auth.policies must contain exactly openbao.policy_name")
        claims = auth.get("bound_claims") or {}
        for key in ("service_account_names", "service_account_namespaces"):
            values = claims.get(key)
            if not (isinstance(values, list) and len(values) == 1 and NAME_RE.match(str(values[0]))):
                errors.append(f"openbao.auth.bound_claims.{key} must name exactly one value")
    admission = ccr.get("admission")
    if not isinstance(admission, dict) or not admission.get("standard") or not admission.get("executor_identity"):
        errors.append("admission must name the admission standard and executor identity")
    if (ccr.get("access_frontdoor") or {}).get("resolvable") is not False:
        errors.append("runtime credential frontdoor must remain disabled")
    if (ccr.get("agent_fence") or {}).get("data_and_metadata_denied") is not True:
        errors.append("coding agents must be fenced from the data and metadata paths")
    if ccr.get("status") in APPROVED:
        approved = _approvals(ccr)
        for role in (ccr.get("review") or {}).get("required_approvers") or []:
            if role not in approved:
                errors.append("explicit role approval missing: " + role)
    for section, keys in {"verification": ["positive", "negative", "activation_conditions"],
                          "lifecycle": ["deactivate", "rotate", "compromised"]}.items():
        for key in keys:
            if not (ccr.get(section) or {}).get(key):
                errors.append(f"{section}.{key} required")
    return errors


def procedure_note(ccr: dict[str, Any]) -> str:
    openbao = ccr["openbao"]
    return (f"Attended platform-admin applies policy {openbao['policy_name']} and Kubernetes role "
            f"{openbao['auth']['role']} from a reviewed commit, sets cas_required on {openbao['kv_path']} "
            f"and proves the negatives ({DOC}). The workload may only write the next version with CAS.")