"""Fixed mutation shapes for the trusted Git/ArgoCD release adapter. These builders confer no authority. The admitted adapter must authenticate its transport, enforce compare-and-swap, and never accept arbitrary path/body inputs. """ import copy import re import yaml REVISION = re.compile(r'[0-9a-f]{40}') APPLICATION_PATH = 'argocd/railiance01/applications/activity-core.application.yaml' ROOT_APPLICATION = 'railiance-apps-root' APPLICATION = 'activity-core' NAMESPACE = 'argocd' def revision(value): if not isinstance(value,str) or REVISION.fullmatch(value) is None: raise ValueError('full commit revision required') return value def update_child(document, expected, target): """Change only targetRevision in the one validated child Application.""" revision(expected);revision(target) before = yaml.safe_load(document) if not isinstance(before,dict):raise ValueError('invalid child declaration') spec=before.get('spec',{});source=spec.get('source',{}) if (before.get('apiVersion')!='argoproj.io/v1alpha1' or before.get('kind')!='Application' or before.get('metadata',{}).get('name')!=APPLICATION or before.get('metadata',{}).get('namespace')!=NAMESPACE or spec.get('project')!='activity-core' or source.get('repoURL')!='https://forgejo.coulomb.social/coulomb/activity-core.git' or source.get('path')!='k8s/gitops' or spec.get('destination')!={'namespace':'activity-core','server':'https://kubernetes.default.svc'} or spec.get('syncPolicy',{}).get('automated') is not None or set(spec.get('syncPolicy',{}).get('syncOptions',[])) != {'CreateNamespace=false','ApplyOutOfSyncOnly=true','PruneLast=true','FailOnSharedResource=true'} or 'sources' in spec or before.get('metadata',{}).get('finalizers')): raise ValueError('child outside release contract') if source.get('targetRevision') not in {expected,target}: raise ValueError('child revision CAS conflict') if source['targetRevision']==target:return document # Preserve comments and all other bytes; ambiguous duplicate YAML keys cannot # silently widen this one-field text patch. pattern=re.compile(r'(?m)^ targetRevision: '+re.escape(expected)+r'[ \t]*$') if len(pattern.findall(document))!=1:raise ValueError('ambiguous child revision field') updated=pattern.sub(' targetRevision: '+target,document) wanted=copy.deepcopy(before);wanted['spec']['source']['targetRevision']=target if yaml.safe_load(updated)!=wanted:raise ValueError('unexpected child mutation') return updated def sync_operations(platform_revision, activity_revision): """Return fixed root-selective and child sync bodies, never prune/overrides.""" revision(platform_revision);revision(activity_revision) def body(value,resources=None): sync={'revision':value,'prune':False,'syncStrategy':{'apply':{}}} if resources is not None:sync['resources']=resources return {'operation':{'initiatedBy':{'username':'activity-core-release-broker'},'sync':sync}} return [ (ROOT_APPLICATION,body(platform_revision,[{'group':'argoproj.io','kind':'Application','name':APPLICATION,'namespace':NAMESPACE}])), (APPLICATION,body(activity_revision)), ]