RPF-WP-0069: answer NK-WP-0056-T02; add workload-kv-write-next-version; CCR-2026-0058 (proposed)

Owner answer for the bounded executor (hub 9d08288d): one Kubernetes role per
lane for sso/nk-bind-executor; new contract-only request type for a CAS
next-version write (no create, metadata or delete, 15m, agents fenced);
CCR-2026-0058 drafted as proposed, not approved; audit sender in the
CCR-2026-0021/0022 shape once audit-core admits it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 994874@bnt-lap001
Assistant-Session: 6017c416-d619-47eb-bd1a-f670a02d7d87

scripts/workload_kv_write_next_version.py

@@ -0,0 +1,119 @@
+#!/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."""
+    openbao = ccr["openbao"]
+    suffix = openbao["kv_path"][len("platform/"):]
+    return (f'path "platform/data/{suffix}" {{\n'
+            '  capabilities = ["read", "update"]\n'
+            '  required_parameters = ["options"]\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.")