tegwick
c12a8fbcfb
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e241-8285-7a63-8c0c-51c9cb824dc3
63 lines
3.6 KiB
Python
63 lines
3.6 KiB
Python
"""Fail-closed admission check for a candidate image-only GitOps release.
|
|
|
|
This is a validator, not an authority issuer or a cluster/Git credential broker.
|
|
Evidence must be supplied by the separately admitted release identity.
|
|
"""
|
|
import argparse
|
|
import copy
|
|
from datetime import datetime, timedelta, timezone
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import yaml
|
|
IMAGE=re.compile(r'forgejo\.coulomb\.social/coulomb/activity-core@sha256:[0-9a-f]{64}')
|
|
SHA=re.compile(r'[0-9a-f]{40}')
|
|
DEPLOYS={'actcore-api','actcore-worker','actcore-event-router'}
|
|
def require(condition,message):
|
|
if not condition: raise ValueError(message)
|
|
def indexed(docs):
|
|
out={}
|
|
for d in docs:
|
|
require(isinstance(d,dict),'invalid resource')
|
|
key=(d['kind'],d['metadata']['name'])
|
|
require(key not in out and d['metadata'].get('namespace')=='activity-core','duplicate or foreign resource')
|
|
out[key]=copy.deepcopy(d)
|
|
return out
|
|
|
|
def validate(before,after,evidence,now=None):
|
|
now=now or datetime.now(timezone.utc)
|
|
require(evidence.get('schema_version')==1,'unknown evidence schema')
|
|
require(evidence.get('identity_admitted') is True,'release identity not admitted')
|
|
require(evidence.get('authority')=='ACTIVITY-WP-0041-image-only-v1','unknown authority')
|
|
require(evidence.get('review_result')=='pass' and evidence.get('checks')=='pass','checks/review not passed')
|
|
require(evidence.get('reviewer') and evidence.get('reviewer')!=evidence.get('producer'),'independent review required')
|
|
require(bool(SHA.fullmatch(evidence.get('candidate_commit',''))),'full candidate revision required')
|
|
require(bool(SHA.fullmatch(evidence.get('rollback_commit',''))),'rollback revision required')
|
|
require(evidence['candidate_commit']!=evidence['rollback_commit'],'rollback must name prior revision')
|
|
observed=datetime.fromisoformat(evidence['healthy_since'].replace('Z','+00:00'))
|
|
measured=datetime.fromisoformat(evidence['observed_at'].replace('Z','+00:00'))
|
|
require(observed.tzinfo is not None and measured.tzinfo is not None,'timestamps need timezone')
|
|
require(now-timedelta(minutes=5)<=measured<=now,'health evidence stale or future')
|
|
require(measured-observed>=timedelta(hours=24),'24-hour healthy observation period incomplete')
|
|
require(evidence.get('argo_synced') is True and evidence.get('argo_healthy') is True,'Argo not healthy/synced')
|
|
left,right=indexed(before),indexed(after)
|
|
require(left.keys()==right.keys(),'resource inventory change')
|
|
changed=[]
|
|
for key,a in left.items():
|
|
b=right[key]
|
|
if a==b: continue
|
|
require(key[0]=='Deployment' and key[1] in DEPLOYS,'only runtime images may change')
|
|
ac=a['spec']['template']['spec']['containers'];bc=b['spec']['template']['spec']['containers']
|
|
require(len(ac)==len(bc)==1,'container inventory change')
|
|
require(bool(IMAGE.fullmatch(bc[0].get('image',''))),'registry digest required')
|
|
require(bc[0].get('imagePullPolicy')=='IfNotPresent','pull policy must support registry recovery')
|
|
ac[0]['image']=bc[0]['image'];ac[0]['imagePullPolicy']=bc[0]['imagePullPolicy']
|
|
require(a==b,'non-image deployment change')
|
|
changed.append(key[1])
|
|
require(bool(changed),'no release change')
|
|
return {'admitted':True,'deployments':changed,'candidate_commit':evidence['candidate_commit'],'rollback_commit':evidence['rollback_commit']}
|
|
|
|
if __name__=='__main__':
|
|
p=argparse.ArgumentParser();p.add_argument('before',type=Path);p.add_argument('after',type=Path);p.add_argument('evidence',type=Path);a=p.parse_args()
|
|
try: print(json.dumps(validate(list(yaml.safe_load_all(a.before.read_text())),list(yaml.safe_load_all(a.after.read_text())),json.loads(a.evidence.read_text()))))
|
|
except (ValueError,KeyError,TypeError) as e: raise SystemExit(f'refused: {e}')
|