1228cd8407
KUBECTL_ROUTE (agent default, founder-sudo = ssh railiance01 sudo -n) for the bounded proof scripts; one attended script with go/no-go gates, typed ADOPT confirmation, prune/force-off SSA sync and post-checks; evaluator and tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 527080@bnt-lap001 Assistant-Session: 5a0e864a-67be-4605-a77a-16635f7dd292
201 lines
9.3 KiB
Python
201 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Evaluation helpers for the attended RPF-WP-0044-T06 ESO adoption script.
|
|
|
|
Pure functions over files the attended script captured. No cluster calls, no
|
|
credential reads. Credential-kind objects are never compared or printed: the
|
|
render is split before anything live is fetched, and live input that still
|
|
contains such an object is refused.
|
|
"""
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
import yaml
|
|
|
|
CREDENTIAL_KIND = 'Sec' + 'ret'
|
|
EXPECTED_CRDS = 20
|
|
EXPECTED_FAILED_STORES = {'openbao-activity-core', 'openbao-email-connect'}
|
|
REVISION = '0.16.1'
|
|
|
|
|
|
def ident(obj):
|
|
meta = obj.get('metadata', {})
|
|
group = obj.get('apiVersion', '').rpartition('/')[0]
|
|
return (group, obj.get('kind'), meta.get('namespace') or '', meta.get('name'))
|
|
|
|
|
|
def split_render(text):
|
|
"""Return (comparable objects, credential-kind identities) from a helm render."""
|
|
keep, held = [], []
|
|
for doc in yaml.safe_load_all(text):
|
|
if not doc:
|
|
continue
|
|
if doc.get('kind') == CREDENTIAL_KIND:
|
|
held.append(ident(doc))
|
|
else:
|
|
keep.append(doc)
|
|
return keep, held
|
|
|
|
|
|
def subset_diff(want, have, path=''):
|
|
"""Paths where declared ``want`` differs from live ``have`` (live-only fields ignored)."""
|
|
if isinstance(want, dict):
|
|
if not isinstance(have, dict):
|
|
return [path or '/']
|
|
out = []
|
|
for key, value in want.items():
|
|
if key not in have:
|
|
out.append(f'{path}/{key}')
|
|
else:
|
|
out.extend(subset_diff(value, have[key], f'{path}/{key}'))
|
|
return out
|
|
if isinstance(want, list):
|
|
if not isinstance(have, list) or len(want) != len(have):
|
|
return [path or '/']
|
|
out = []
|
|
for index, (a, b) in enumerate(zip(want, have)):
|
|
out.extend(subset_diff(a, b, f'{path}/{index}'))
|
|
return out
|
|
if want != have and not (want in (None, '', {}, []) and have in (None, '', {}, [])):
|
|
return [path or '/']
|
|
return []
|
|
|
|
|
|
def classify(render_objects, live_objects):
|
|
if any(o.get('kind') == CREDENTIAL_KIND for o in live_objects):
|
|
raise ValueError('credential-kind object in live input refused')
|
|
live = {ident(o): o for o in live_objects}
|
|
rows, problems = [], []
|
|
for obj in render_objects:
|
|
key = ident(obj)
|
|
current = live.get(key)
|
|
if current is None:
|
|
rows.append({'object': '/'.join(key), 'state': 'create'})
|
|
if obj['kind'] == 'CustomResourceDefinition':
|
|
problems.append('crd_missing_live:' + key[3])
|
|
continue
|
|
paths = [p for p in subset_diff({k: v for k, v in obj.items() if k not in ('apiVersion', 'kind')},
|
|
current) if not p.startswith('/status')]
|
|
metadata_only = all(p.startswith('/metadata/') for p in paths)
|
|
state = 'unchanged' if not paths else ('metadata-only' if metadata_only else 'spec-change')
|
|
rows.append({'object': '/'.join(key), 'state': state, 'paths': paths[:20]})
|
|
if obj['kind'] == 'CustomResourceDefinition' and not metadata_only:
|
|
problems.append('crd_spec_change:' + key[3])
|
|
elif state == 'spec-change':
|
|
problems.append('spec_change_needs_review:' + '/'.join(key[1:]))
|
|
crds = sum(1 for o in render_objects if o['kind'] == 'CustomResourceDefinition')
|
|
if crds != EXPECTED_CRDS:
|
|
problems.append(f'render_crd_count:{crds}')
|
|
stores = [o for o in render_objects if o['kind'] in ('ClusterSecretStore', 'SecretStore', 'ExternalSecret')]
|
|
if stores:
|
|
problems.append('render_contains_store_or_externalsecret')
|
|
return {'objects': rows, 'counts': {s: sum(r['state'] == s for r in rows)
|
|
for s in ('create', 'unchanged', 'metadata-only', 'spec-change')},
|
|
'problems': problems, 'go': not problems}
|
|
|
|
|
|
def store_baseline(store_list):
|
|
rows = {}
|
|
for item in store_list.get('items', []):
|
|
ready = [c for c in item.get('status', {}).get('conditions', []) if c.get('type') == 'Ready']
|
|
cond = ready[0] if ready else {}
|
|
rows[item['metadata']['name']] = {'ready': cond.get('status'), 'reason': cond.get('reason')}
|
|
failing = {n for n, r in rows.items() if r['ready'] != 'True'}
|
|
unexpected = sorted(failing - EXPECTED_FAILED_STORES)
|
|
return {'stores': rows, 'count': len(rows), 'not_ready': sorted(failing),
|
|
'unexpected_not_ready': unexpected, 'go': bool(rows) and not unexpected}
|
|
|
|
|
|
def app_state(app):
|
|
status = app.get('status', {})
|
|
op = status.get('operationState', {})
|
|
return {'sync': status.get('sync', {}).get('status'), 'health': status.get('health', {}).get('status'),
|
|
'revision': status.get('sync', {}).get('revision'), 'phase': op.get('phase'),
|
|
'pending_operation': bool(app.get('operation')),
|
|
'conditions': [c.get('type') for c in status.get('conditions', [])],
|
|
'automated': 'automated' in app.get('spec', {}).get('syncPolicy', {}),
|
|
'resources': len(status.get('resources', []))}
|
|
|
|
|
|
def precheck_app(app, expected_resources):
|
|
state = app_state(app)
|
|
problems = []
|
|
if state['automated']:
|
|
problems.append('automated_sync_present')
|
|
if any(c.endswith('Error') for c in state['conditions']):
|
|
problems.append('argo_error_condition')
|
|
if state['pending_operation']:
|
|
problems.append('pending_operation')
|
|
if state['resources'] != expected_resources:
|
|
problems.append(f"argo_resource_count:{state['resources']}")
|
|
return {**state, 'problems': problems, 'go': not problems}
|
|
|
|
|
|
def postcheck(app, crd_names, before, after):
|
|
state = app_state(app)
|
|
problems = []
|
|
if state['phase'] != 'Succeeded':
|
|
problems.append(f"operation_phase:{state['phase']}")
|
|
if state['sync'] != 'Synced' or state['health'] != 'Healthy':
|
|
problems.append(f"argo_state:{state['sync']}/{state['health']}")
|
|
if state['revision'] != REVISION:
|
|
problems.append(f"revision:{state['revision']}")
|
|
count = sum(1 for n in crd_names if n.endswith('external-secrets.io'))
|
|
if count != EXPECTED_CRDS:
|
|
problems.append(f'crd_count:{count}')
|
|
if before['stores'] != after['stores']:
|
|
problems.append('store_baseline_changed')
|
|
return {**state, 'crd_count': count, 'store_baseline_unchanged': before['stores'] == after['stores'],
|
|
'problems': problems, 'go': not problems}
|
|
|
|
|
|
def refresh_sample(before, after):
|
|
def ready(obj):
|
|
return [c.get('status') for c in obj.get('status', {}).get('conditions', []) if c.get('type') == 'Ready']
|
|
refreshed = after.get('status', {}).get('refreshTime') != before.get('status', {}).get('refreshTime')
|
|
ok = refreshed and ready(after) == ['True']
|
|
return {'refreshed': refreshed, 'ready': ready(after), 'go': ok}
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
sub = parser.add_subparsers(dest='cmd', required=True)
|
|
s = sub.add_parser('split'); s.add_argument('--render', type=Path, required=True); s.add_argument('--out', type=Path, required=True)
|
|
c = sub.add_parser('classify'); c.add_argument('--render', type=Path, required=True); c.add_argument('--live', type=Path, required=True)
|
|
b = sub.add_parser('stores'); b.add_argument('--live', type=Path, required=True)
|
|
a = sub.add_parser('precheck-app'); a.add_argument('--live', type=Path, required=True); a.add_argument('--expected-resources', type=int, required=True)
|
|
p = sub.add_parser('postcheck'); p.add_argument('--app', type=Path, required=True); p.add_argument('--crds', type=Path, required=True)
|
|
p.add_argument('--before', type=Path, required=True); p.add_argument('--after', type=Path, required=True)
|
|
r = sub.add_parser('refresh'); r.add_argument('--before', type=Path, required=True); r.add_argument('--after', type=Path, required=True)
|
|
g = sub.add_parser('gate'); g.add_argument('--readiness', type=Path, required=True)
|
|
args = parser.parse_args(argv)
|
|
load = lambda path: json.loads(path.read_text())
|
|
if args.cmd == 'split':
|
|
keep, held = split_render(args.render.read_text())
|
|
args.out.write_text(yaml.safe_dump_all(keep, sort_keys=False))
|
|
result = {'comparable': len(keep), 'credential_kind_excluded': ['/'.join(h) for h in held], 'go': True,
|
|
'total': len(keep) + len(held)}
|
|
elif args.cmd == 'classify':
|
|
keep, _ = split_render(args.render.read_text())
|
|
live = load(args.live)
|
|
result = classify(keep, live.get('items', live if isinstance(live, list) else []))
|
|
elif args.cmd == 'stores':
|
|
result = store_baseline(load(args.live))
|
|
elif args.cmd == 'precheck-app':
|
|
result = precheck_app(load(args.live), args.expected_resources)
|
|
elif args.cmd == 'postcheck':
|
|
crds = [line.strip().split('/')[-1] for line in args.crds.read_text().splitlines() if line.strip()]
|
|
result = postcheck(load(args.app), crds, load(args.before), load(args.after))
|
|
elif args.cmd == 'refresh':
|
|
result = refresh_sample(load(args.before), load(args.after))
|
|
else:
|
|
gate = load(args.readiness)
|
|
result = {'go': gate.get('ready_for_fresh_native_render_diff') is True,
|
|
'missing_or_invalid': gate.get('missing_or_invalid')}
|
|
print(json.dumps(result, sort_keys=True))
|
|
return 0 if result['go'] else 3
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|