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