Close CCR drift and high-risk policy gaps

scripts/credential-change.py

@@ -22,6 +22,7 @@ REPO_DIR = Path(__file__).resolve().parents[1]
 DEFAULT_CCR_DIR = REPO_DIR / "credential-change-requests"
 ALLOWED_STATUSES = {
     "draft",
+    "in_flight",
     "proposed",
     "needs_changes",
     "approved",
@@ -189,6 +190,28 @@ def reject_secret_text(text: str, field: str) -> None:
 
 
 def validate_workload_kv_read(ccr: dict[str, Any], errors: list[str], warnings: list[str]) -> None:
+    in_flight = ccr.get("status") == "in_flight"
+    missing_fields: set[str] = set()
+    if in_flight:
+        declaration = require_object(ccr.get("in_flight"), "in_flight", errors)
+        listed_missing = require_list(
+            declaration.get("missing_fields"), "in_flight.missing_fields", errors
+        )
+        missing_fields = {str(field) for field in listed_missing}
+        if not missing_fields:
+            errors.append("in_flight.missing_fields must not be empty")
+        allowed_missing = {"openbao.policy_file", "openbao.auth"}
+        unsupported_missing = missing_fields - allowed_missing
+        if unsupported_missing:
+            errors.append(
+                "in_flight.missing_fields contains unsupported fields: "
+                + ", ".join(sorted(unsupported_missing))
+            )
+        require_string(
+            declaration.get("blocking_reason"), "in_flight.blocking_reason", errors
+        )
+        require_string(declaration.get("owner"), "in_flight.owner", errors)
+
     target = require_object(ccr.get("target"), "target", errors)
     for field in ("domain", "tenant", "workload", "environment", "purpose"):
         require_string(target.get(field), f"target.{field}", errors)
@@ -203,9 +226,16 @@ def validate_workload_kv_read(ccr: dict[str, Any], errors: list[str], warnings:
     policy_name = require_string(
         openbao.get("policy_name"), "openbao.policy_name", errors
     )
-    policy_file = require_string(
-        openbao.get("policy_file"), "openbao.policy_file", errors
-    )
+    policy_file = ""
+    if "openbao.policy_file" in missing_fields:
+        if openbao.get("policy_file") is not None:
+            errors.append(
+                "openbao.policy_file is declared missing but is present"
+            )
+    else:
+        policy_file = require_string(
+            openbao.get("policy_file"), "openbao.policy_file", errors
+        )
     fields = [str(field) for field in require_list(openbao.get("fields"), "openbao.fields", errors)]
     if not fields:
         errors.append("openbao.fields must contain at least one field")
@@ -220,12 +250,20 @@ def validate_workload_kv_read(ccr: dict[str, Any], errors: list[str], warnings:
         if not resolved_policy.exists():
             errors.append(f"openbao.policy_file does not exist: {policy_file}")
 
-    auth = require_object(openbao.get("auth"), "openbao.auth", errors)
-    method = require_string(auth.get("method"), "openbao.auth.method", errors)
-    if method not in {"oidc", "kubernetes"}:
-        errors.append("openbao.auth.method must be oidc or kubernetes")
-    require_string(auth.get("mount"), "openbao.auth.mount", errors)
-    require_string(auth.get("role"), "openbao.auth.role", errors)
+    auth: dict[str, Any] = {}
+    if "openbao.auth" in missing_fields:
+        if openbao.get("auth") is not None:
+            errors.append("openbao.auth is declared missing but is present")
+    else:
+        auth = require_object(openbao.get("auth"), "openbao.auth", errors)
+    method = ""
+    if auth:
+        method = require_string(auth.get("method"), "openbao.auth.method", errors)
+    if method and method not in {"oidc", "kubernetes", "token"}:
+        errors.append("openbao.auth.method must be oidc, kubernetes, or token")
+    if method in {"oidc", "kubernetes"}:
+        require_string(auth.get("mount"), "openbao.auth.mount", errors)
+        require_string(auth.get("role"), "openbao.auth.role", errors)
     if method == "oidc":
         redirect_uris = require_list(
             auth.get("allowed_redirect_uris"),
@@ -252,22 +290,95 @@ def validate_workload_kv_read(ccr: dict[str, Any], errors: list[str], warnings:
                     errors.append(
                         f"openbao.auth.oidc_scopes[{index}] must be a non-empty string"
                     )
-    policies = [str(policy) for policy in require_list(auth.get("policies"), "openbao.auth.policies", errors)]
-    if policies != [policy_name]:
-        errors.append("openbao.auth.policies must contain exactly openbao.policy_name")
-    for policy in policies:
-        if policy in DISALLOWED_POLICY_NAMES:
-            errors.append(f"openbao.auth.policies contains disallowed policy {policy}")
-    ttl = auth.get("ttl")
-    if ttl is not None and (not isinstance(ttl, str) or not TTL_RE.match(ttl)):
-        errors.append("openbao.auth.ttl must match <positive integer><s|m|h|d>")
-    bound_claims = require_object(
-        auth.get("bound_claims"), "openbao.auth.bound_claims", errors
-    )
-    if not bound_claims:
-        errors.append("openbao.auth.bound_claims must not be empty")
-    if auth.get("bound_claims_confirmed") is not True:
-        warnings.append("OIDC/Kubernetes bound claim is not confirmed; apply is blocked")
+    policies: list[str] = []
+    if auth:
+        policies = [
+            str(policy)
+            for policy in require_list(
+                auth.get("policies"), "openbao.auth.policies", errors
+            )
+        ]
+        expected_policy = policy_name
+        if method == "token":
+            expected_policy = require_string(
+                openbao.get("eso_policy_name"), "openbao.eso_policy_name", errors
+            )
+            eso_policy_file = require_string(
+                openbao.get("eso_policy_file"), "openbao.eso_policy_file", errors
+            )
+            if eso_policy_file and not resolve_repo_path(eso_policy_file).exists():
+                errors.append(
+                    f"openbao.eso_policy_file does not exist: {eso_policy_file}"
+                )
+        if policies != [expected_policy]:
+            errors.append(
+                "openbao.auth.policies must contain exactly the policy used by the auth method"
+            )
+        for policy in policies:
+            if policy in DISALLOWED_POLICY_NAMES:
+                errors.append(
+                    f"openbao.auth.policies contains disallowed policy {policy}"
+                )
+        ttl = auth.get("ttl")
+        if ttl is not None and (
+            not isinstance(ttl, str) or not TTL_RE.match(ttl)
+        ):
+            errors.append("openbao.auth.ttl must match <positive integer><s|m|h|d>")
+    if method in {"oidc", "kubernetes"}:
+        bound_claims = require_object(
+            auth.get("bound_claims"), "openbao.auth.bound_claims", errors
+        )
+        if not bound_claims:
+            errors.append("openbao.auth.bound_claims must not be empty")
+        if auth.get("bound_claims_confirmed") is not True:
+            warnings.append(
+                "OIDC/Kubernetes bound claim is not confirmed; apply is blocked"
+            )
+    elif method == "token":
+        token_secret = require_string(
+            auth.get("token_secret"), "openbao.auth.token_secret", errors
+        )
+        if token_secret and not re.match(
+            r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?/[a-z0-9]([-a-z0-9]*[a-z0-9])?$",
+            token_secret,
+        ):
+            errors.append("openbao.auth.token_secret must be namespace/name")
+        require_string(
+            auth.get("bootstrap_script"), "openbao.auth.bootstrap_script", errors
+        )
+        if auth.get("ttl") is None:
+            errors.append("openbao.auth.ttl is required for token auth")
+        followup = require_object(
+            auth.get("kubernetes_followup"),
+            "openbao.auth.kubernetes_followup",
+            errors,
+        )
+        if followup.get("method") != "kubernetes":
+            errors.append(
+                "openbao.auth.kubernetes_followup.method must be kubernetes"
+            )
+        require_string(
+            followup.get("mount"),
+            "openbao.auth.kubernetes_followup.mount",
+            errors,
+        )
+        require_string(
+            followup.get("role"),
+            "openbao.auth.kubernetes_followup.role",
+            errors,
+        )
+        followup_claims = require_object(
+            followup.get("bound_claims"),
+            "openbao.auth.kubernetes_followup.bound_claims",
+            errors,
+        )
+        if not followup_claims:
+            errors.append(
+                "openbao.auth.kubernetes_followup.bound_claims must not be empty"
+            )
+        warnings.append(
+            "token auth is transitional; complete the declared Kubernetes-auth follow-up"
+        )
 
     frontdoor = require_object(ccr.get("access_frontdoor"), "access_frontdoor", errors)
     require_string(frontdoor.get("type"), "access_frontdoor.type", errors)
@@ -342,14 +453,11 @@ def validate_ccr(path: Path) -> tuple[dict[str, Any], list[str], list[str]]:
 
 def render_summary(ccr: dict[str, Any], warnings: list[str]) -> str:
     openbao = ccr["openbao"]
-    auth = openbao["auth"]
+    auth = openbao.get("auth") or {}
     frontdoor = ccr["access_frontdoor"]
     risk = ccr["risk"]
     verification = ccr["verification"]
     fields = ", ".join(openbao["fields"])
-    claim_bits = ", ".join(
-        f"{key}={value}" for key, value in auth.get("bound_claims", {}).items()
-    )
     lines = [
         f"Request: {ccr['title']}",
         f"CCR: {ccr['id']} ({ccr['status']})",
@@ -361,13 +469,37 @@ def render_summary(ccr: dict[str, Any], warnings: list[str]) -> str:
         "Policy:",
         f"  {openbao['policy_name']}",
         "Auth binding:",
-        f"  {auth['mount']} {auth['method']} role {auth['role']}",
-        f"  bound claims: {claim_bits}",
-        f"  confirmed: {auth.get('bound_claims_confirmed') is True}",
-        "Access front door:",
-        f"  {frontdoor['type']} {frontdoor['catalog_id']}",
-        f"  readiness: {frontdoor.get('readiness')} resolvable={frontdoor.get('resolvable') is True}",
     ]
+    if auth.get("method") == "token":
+        lines.extend(
+            [
+                f"  transitional token via Secret {auth['token_secret']}",
+                f"  policy: {', '.join(auth.get('policies', []))}",
+                f"  ttl: {auth.get('ttl')}",
+            ]
+        )
+    elif auth:
+        claim_bits = ", ".join(
+            f"{key}={value}"
+            for key, value in auth.get("bound_claims", {}).items()
+        )
+        lines.extend(
+            [
+                f"  {auth['mount']} {auth['method']} role {auth['role']}",
+                f"  bound claims: {claim_bits}",
+                f"  confirmed: {auth.get('bound_claims_confirmed') is True}",
+            ]
+        )
+    else:
+        missing = ", ".join(ccr.get("in_flight", {}).get("missing_fields", []))
+        lines.append(f"  in flight; declared missing: {missing}")
+    lines.extend(
+        [
+            "Access front door:",
+            f"  {frontdoor['type']} {frontdoor['catalog_id']}",
+            f"  readiness: {frontdoor.get('readiness')} resolvable={frontdoor.get('resolvable') is True}",
+        ]
+    )
     if frontdoor.get("command"):
         lines.append(f"  command: {frontdoor['command']}")
     lines.append(f"Risk: {risk['classification']}")
@@ -1408,8 +1540,15 @@ def apply_blockers(ccr: dict[str, Any]) -> list[str]:
         return blockers
     if status not in APPLY_ALLOWED_STATUSES:
         blockers.append(f"apply requires status approved, got {status}")
-    if ccr["openbao"]["auth"].get("bound_claims_confirmed") is not True:
+    auth = ccr["openbao"].get("auth") or {}
+    if auth.get("method") in {"oidc", "kubernetes"} and auth.get(
+        "bound_claims_confirmed"
+    ) is not True:
         blockers.append("apply requires confirmed OpenBao auth binding")
+    if auth.get("method") == "token":
+        blockers.append(
+            "delegated apply does not create transitional token-auth bootstrap identities"
+        )
     return blockers
 
 
@@ -1432,7 +1571,7 @@ def status_payload(ccr: dict[str, Any], warnings: list[str]) -> dict[str, Any]:
     frontdoor_blocked_by = frontdoor_blockers(ccr)
     frontdoor = ccr["access_frontdoor"]
     openbao = ccr["openbao"]
-    auth = openbao["auth"]
+    auth = openbao.get("auth") or {}
     return {
         "id": ccr["id"],
         "title": ccr["title"],
@@ -1449,9 +1588,10 @@ def status_payload(ccr: dict[str, Any], warnings: list[str]) -> dict[str, Any]:
             "kv_path": openbao["kv_path"],
             "fields": openbao["fields"],
             "policy_name": openbao["policy_name"],
-            "auth_mount": auth["mount"],
-            "auth_method": auth["method"],
-            "auth_role": auth["role"],
+            "auth_mount": auth.get("mount"),
+            "auth_method": auth.get("method"),
+            "auth_role": auth.get("role"),
+            "token_secret": auth.get("token_secret"),
             "bound_claims_confirmed": auth.get("bound_claims_confirmed") is True,
         },
         "access_frontdoor": {