tegwick
994d11c869
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
65 lines
3.6 KiB
Python
65 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Plan two exact DNS-selector exclusions; apply only an owner-admitted hash-bound plan."""
|
|
import argparse
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
HERE=Path(__file__).resolve().parent
|
|
spec=importlib.util.spec_from_file_location('capture',HERE/'capture-identity.py')
|
|
capture=importlib.util.module_from_spec(spec);spec.loader.exec_module(capture)
|
|
SELECTOR={'matchExpressions':[{'key':'net-kingdom/capture','operator':'DoesNotExist'}]}
|
|
|
|
|
|
def plan():
|
|
if capture.get(None,'namespace','kube-system')['metadata']['uid']!=capture.prepare.CLUSTER_UID:raise capture.Refused()
|
|
rows=[]
|
|
for ns in ('sso','mfa'):
|
|
obj=capture.get(ns,'networkpolicy','allow-egress-dns');s=obj['spec'];m=obj['metadata']
|
|
if s['podSelector'] not in ({},SELECTOR):raise capture.Refused()
|
|
pods=json.loads(capture.command(['-n',ns,'get','pods','-o','json']))['items']
|
|
if any('net-kingdom/capture' in p['metadata'].get('labels',{}) for p in pods):raise capture.Refused()
|
|
# Preserve every peer, port and policy type. Existing unlabelled service
|
|
# pods retain their exact selector match. UID/RV/spec tests guard drift.
|
|
patch=[{'op':'test','path':'/metadata/uid','value':m['uid']},
|
|
{'op':'test','path':'/metadata/resourceVersion','value':m['resourceVersion']},
|
|
{'op':'test','path':'/spec','value':s},
|
|
{'op':'replace','path':'/spec/podSelector','value':SELECTOR}]
|
|
rows.append({'namespace':ns,'name':m['name'],'uid':m['uid'],'patch':patch})
|
|
return {'schema':'netkingdom.identity-isolation.v1','status':'prepared_not_applied',
|
|
'kubeconfig':capture.KUBE[2],'source_sha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),'policies':rows}
|
|
|
|
|
|
def apply(packet,receipt,dry_run=False):
|
|
if packet!=plan():raise capture.Refused()
|
|
for row in packet['policies']:
|
|
args=['-n',row['namespace'],'patch','networkpolicy',row['name'],'--type=json','-p',json.dumps(row['patch']),'-o','json']
|
|
if dry_run:args+=['--dry-run=server']
|
|
receipt['attempted'].append({'namespace':row['namespace'],'name':row['name'],'uid':row['uid']})
|
|
actual=json.loads(capture.command(args));expected=row['patch'][2]['value'].copy();expected['podSelector']=SELECTOR
|
|
if actual['metadata']['uid']!=row['uid'] or actual['spec']!=expected:raise capture.Refused()
|
|
receipt['verified'].append({'namespace':row['namespace'],'name':row['name'],'uid':row['uid']})
|
|
receipt['status']='server_dry_run_passed' if dry_run else 'capture_dns_exclusions_verified'
|
|
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser(description=__doc__);p.add_argument('action',choices=['plan','dry-run','apply'])
|
|
p.add_argument('--kubeconfig',required=True);p.add_argument('--plan');p.add_argument('--plan-sha256');p.add_argument('--receipt',required=True)
|
|
a=p.parse_args();os.umask(0o077);capture.KUBE=['kubectl','--kubeconfig',a.kubeconfig]
|
|
fd=os.open(a.receipt,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
|
|
r={'status':'failed','attempted':[],'verified':[]}
|
|
try:
|
|
if a.action=='plan':r=plan()
|
|
else:
|
|
raw=Path(a.plan).read_bytes()
|
|
if hashlib.sha256(raw).hexdigest()!=a.plan_sha256:raise capture.Refused()
|
|
apply(json.loads(raw),r,a.action=='dry-run')
|
|
except Exception:r.update(status='failed',error='isolation_refused_review_partial_attempts')
|
|
finally:
|
|
with os.fdopen(fd,'w') as out:json.dump(r,out,indent=2);out.write('\n')
|
|
return int(r['status']=='failed')
|
|
|
|
|
|
if __name__=='__main__':raise SystemExit(main())
|