Implement signed release admission and durable rollback coordinator

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e241-8285-7a63-8c0c-51c9cb824dc3

scripts/check_gitops_promotion.py

@@ -1,63 +1,19 @@
-"""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.
-"""
+"""Compatibility CLI for the shared image-only release admission policy."""
 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
+from activity_core.gitops_policy import validate
 
-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}')
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("before", type=Path)
+    parser.add_argument("after", type=Path)
+    parser.add_argument("evidence", type=Path)
+    args = parser.parse_args()
+    try:
+        print(json.dumps(validate(list(yaml.safe_load_all(args.before.read_text())),
+                                  list(yaml.safe_load_all(args.after.read_text())),
+                                  json.loads(args.evidence.read_text()))))
+    except (ValueError, KeyError, TypeError) as exc:
+        raise SystemExit(f"refused: {exc}")