Allow bounded ESO token self-validation and cleanup

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883

credential-change-requests/CCR-2026-0015-state-hub-preflight-signing.yaml

@@ -49,6 +49,7 @@ openbao:
   fields:
   - REPOSITORY_RENAME_PREFLIGHT_SECRET
   metadata_read: false
+  token_self_lifecycle: true
   policy_name: workload-kv-read-state-hub-rename-preflight
   policy_file: openbao/policies/workload-kv-read-state-hub-rename-preflight.hcl
   auth:

openbao/policies/workload-kv-read-state-hub-rename-preflight.hcl

@@ -1,3 +1,11 @@
 path "platform/data/workloads/state-hub/repository-rename-preflight" {
   capabilities = ["read"]
 }
+
+path "auth/token/lookup-self" {
+  capabilities = ["read"]
+}
+
+path "auth/token/revoke-self" {
+  capabilities = ["update"]
+}

scripts/credential-change.py

@@ -238,6 +238,8 @@ def validate_workload_kv_read(ccr: dict[str, Any], errors: list[str], warnings:
         )
     if "metadata_read" in openbao and not isinstance(openbao["metadata_read"], bool):
         errors.append("openbao.metadata_read must be boolean")
+    if "token_self_lifecycle" in openbao and not isinstance(openbao["token_self_lifecycle"], bool):
+        errors.append("openbao.token_self_lifecycle must be boolean")
     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")
@@ -544,6 +546,9 @@ def generated_policy_hcl(ccr: dict[str, Any]) -> str:
     if openbao.get("metadata_read", True):
         body += (f'\npath "{mount}/metadata/{suffix}" {{\n'
                  '  capabilities = ["read"]\n' "}\n")
+    if openbao.get("token_self_lifecycle", False):
+        body += ('\npath "auth/token/lookup-self" {\n  capabilities = ["read"]\n}\n'
+                 '\npath "auth/token/revoke-self" {\n  capabilities = ["update"]\n}\n')
     return body
 
 

scripts/state_hub_preflight_lane.py

@@ -61,7 +61,8 @@ def approved_contract():
     if (not auth['bound_claims_confirmed'] or ccr['openbao']['kv_path'] != KV
             or ccr['openbao']['fields'] != [FIELD] or auth['role'] != ROLE
             or auth['mount'] != 'kubernetes' or module.auth_payload(ccr) != expected
-            or ccr['openbao'].get('metadata_read') is not False):
+            or ccr['openbao'].get('metadata_read') is not False
+            or ccr['openbao'].get('token_self_lifecycle') is not True):
         raise LaneError('exact_contract_required')
     policy = module.generated_policy_hcl(ccr)
     if (ROOT / ccr['openbao']['policy_file']).read_text() != policy:
@@ -159,6 +160,15 @@ def run(args, receipt):
     identity = data(bao(['token', 'lookup', '-format=json']))['data']
     if 'platform-admin' not in identity['policies'] or 'root' in identity['policies']:
         raise LaneError('attended_platform_admin_required')
+    if args.action == 'repair-policy':
+        current = data(bao(['read', '-format=json', 'sys/policies/acl/' + POLICY]))['data']['policy']
+        original = policy.split('\npath "auth/token/lookup-self"')[0]
+        if current not in {original, policy}:
+            raise LaneError('read_policy_drift')
+        bao(['write', 'sys/policies/acl/' + POLICY, '-'], payload={'policy': policy})
+        verify_access(kube, receipt)
+        receipt['status'] = 'custody_verified_pending_eso_and_api_acceptance'
+        return
     if args.action == 'verify':
         verify_access(kube, receipt)
         receipt['status'] = 'custody_verified_pending_eso_and_api_acceptance'
@@ -198,7 +208,7 @@ def run(args, receipt):
 
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument('action', choices=['provision', 'rotate', 'verify'])
+    parser.add_argument('action', choices=['provision', 'rotate', 'verify', 'repair-policy'])
     parser.add_argument('--expected-version', required=True, type=int)
     parser.add_argument('--kubeconfig', required=True)
     parser.add_argument('--receipt', required=True)

tests/test_state_hub_preflight_lane.py

@@ -24,6 +24,8 @@ class SigningLaneTests(unittest.TestCase):
 
     def test_data_only_policy_and_bounded_auth_survive_plan_generation(self):
         self.assertNotIn('/metadata/', cc.generated_policy_hcl(self.ccr))
+        self.assertIn('path "auth/token/lookup-self"', cc.generated_policy_hcl(self.ccr))
+        self.assertIn('path "auth/token/revoke-self"', cc.generated_policy_hcl(self.ccr))
         auth = cc.auth_payload(self.ccr)
         self.assertEqual(auth['audience'], 'openbao')
         self.assertEqual(auth['token_explicit_max_ttl'], '15m')
@@ -35,6 +37,7 @@ class SigningLaneTests(unittest.TestCase):
 
     def test_malformed_security_options_rejected(self):
         for section, key, value in [('openbao', 'metadata_read', 'false'),
+                                     ('openbao', 'token_self_lifecycle', 'true'),
                                      ('auth', 'audience', ''),
                                      ('auth', 'token_max_ttl', 900),
                                      ('auth', 'token_no_default_policy', 'true')]: