Record native ESO result and prepare scoped self-token amendment
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a10780-6d41-7952-9718-dfbc1f41236f
This commit is contained in:
tegwick 2026-10-04 23:21:55 +02:00
parent f7776013af
commit e84b68db01
7 changed files with 111 additions and 41 deletions

View file

@ -17,6 +17,8 @@ ROOT = Path(__file__).resolve().parents[1]
CCR = ROOT / 'credential-change-requests/CCR-2026-0032.yaml'
POLICY = 'workload-kv-read-informed-decision-sitting-requester-eso'
ROLE = 'auth/kubernetes/role/informed-decision-sitting-requester-eso'
LEGACY_POLICY = 'path "platform/data/workloads/informed-decision/sitting-requester" {\n capabilities = ["read"]\n}\n'
EXPECTED_POLICY = 'path "platform/data/workloads/informed-decision/sitting-requester" {\n capabilities = ["read"]\n}\n\npath "auth/token/lookup-self" {\n capabilities = ["read"]\n}\n\npath "auth/token/revoke-self" {\n capabilities = ["update"]\n}\n'
class Refused(Exception):
@ -75,7 +77,7 @@ def source(expected_digest):
auth = bao['auth']
if (bao['policy_name'] != POLICY or bao['kv_path'] != 'platform/workloads/informed-decision/sitting-requester'
or bao['fields'] != ['CLIENT_SECRET'] or bao.get('metadata_read') is not False
or bao.get('token_self_lifecycle', False) is not False
or bao.get('token_self_lifecycle', False) is not True
or auth['method'] != 'kubernetes' or auth['mount'] != 'kubernetes'
or auth['role'] != ROLE.rsplit('/', 1)[1]):
raise Refused(2)
@ -87,7 +89,7 @@ def source(expected_digest):
'token_no_default_policy': True}:
raise Refused(2)
policy = module.generated_policy_hcl(ccr)
if (ROOT / bao['policy_file']).read_text() != policy:
if policy != EXPECTED_POLICY or (ROOT / bao['policy_file']).read_text() != policy:
raise Refused(2)
return policy, payload
@ -98,11 +100,13 @@ def apply(policy, payload, *, command=cli, phase=lambda value: None):
phase('read_role')
current_role = command(['read', '-format=json', ROLE], missing_ok=True)
# Check BOTH before mutating either; differing objects require a fresh review.
if current_policy is not None and current_policy.get('data', {}).get('policy') != policy:
existing_policy = current_policy.get('data', {}).get('policy') if current_policy is not None else None
allowed_predecessor = policy == EXPECTED_POLICY and existing_policy == LEGACY_POLICY
if current_policy is not None and existing_policy != policy and not allowed_predecessor:
raise Refused(3)
if current_role is not None and not role_matches(current_role.get('data')):
raise Refused(3)
if current_policy is None:
if current_policy is None or allowed_predecessor:
phase('write_policy')
command(['write', '-format=json', 'sys/policies/acl/' + POLICY, '-'],
json.dumps({'policy': policy}).encode())