64 lines
3.5 KiB
Python
64 lines
3.5 KiB
Python
import importlib.util,json,unittest
|
|
from pathlib import Path
|
|
P=Path(__file__).parent
|
|
spec=importlib.util.spec_from_file_location('preflight',P/'preflight.py');m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
|
|
payload=json.loads((P/'payload.json').read_text())
|
|
class Checks(unittest.TestCase):
|
|
def setUp(self):self.original=m.invoke
|
|
def tearDown(self):m.invoke=self.original
|
|
def test_explicit_fields_and_shape(self):
|
|
calls=[]
|
|
def fake(argv,stdin=None):calls.append(argv);return 0,'alpha\x1e["beta"]\n',False
|
|
m.invoke=fake
|
|
result=m.query('pods',['metadata.name','spec.containers[*].image'],namespace='flex-auth')
|
|
self.assertEqual(result['rows'],[{'metadata.name':'alpha','spec.containers[*].image':['beta']}])
|
|
self.assertIn('-o',calls[0]);self.assertTrue(calls[0][-1].startswith('jsonpath='))
|
|
def test_error_suppression(self):
|
|
m.invoke=lambda *a:(1,'SENTINEL_SECRET',True)
|
|
self.assertNotIn('SENTINEL_SECRET',json.dumps(m.query('pods',['metadata.name'])))
|
|
def test_fail_closed_identity(self):
|
|
m.invoke=lambda *a:(1,'SENTINEL_SECRET',True)
|
|
result=m.run(payload)
|
|
self.assertEqual(result['refused'],'target-identity');self.assertNotIn('SENTINEL_SECRET',json.dumps(result))
|
|
def test_exact_address_and_cluster_shape(self):
|
|
original_query=m.query
|
|
for addresses,cluster in [([{'type':'Hostname','address':'92.205.62.239'}],[{'metadata.uid':'a553c742-0115-43d4-99a4-a5ca56fe0786'}]),([{'type':'InternalIP','address':'192.205.62.239'}],[{'metadata.uid':'a553c742-0115-43d4-99a4-a5ca56fe0786'}]),([{'type':'InternalIP','address':'92.205.62.239'}],[])]:
|
|
calls=[]
|
|
def query(resource,fields,**kw):
|
|
calls.append(resource)
|
|
return {'ok':True,'rows':[{'status.addresses':addresses}]} if resource=='nodes' else {'ok':True,'rows':cluster}
|
|
m.query=query
|
|
try:
|
|
result=m.run(payload);self.assertEqual(result['refused'],'target-identity');self.assertEqual(calls,['nodes','namespace'])
|
|
finally:m.query=original_query
|
|
def test_exact_projection(self):
|
|
deploy=next(x for x in payload['resources'] if x['kind']=='Deployment')
|
|
self.assertEqual(m.deployment_projection(deploy),payload['expected_projection'])
|
|
def test_mock_admission_no_secret_error_leak(self):
|
|
original_query=m.query
|
|
def query(resource,fields,**kw):
|
|
if resource=='nodes':return {'ok':True,'rows':[{'status.addresses':[{'type':'InternalIP','address':'92.205.62.239'}]}]}
|
|
if resource=='namespace':return {'ok':True,'rows':[{'metadata.uid':'a553c742-0115-43d4-99a4-a5ca56fe0786'}]}
|
|
return {'ok':True,'rows':[]}
|
|
m.query=query
|
|
calls=[]
|
|
def fake(argv,stdin=None):
|
|
calls.append(argv)
|
|
if argv[0]=='helm':return 0,'[]',False
|
|
if '--dry-run=server' in argv:
|
|
obj=json.loads(stdin)
|
|
if obj['kind']=='Deployment':
|
|
values=list(m.deployment_projection(obj).values())+[obj['spec']['replicas']]
|
|
return 0,'\x1e'.join(json.dumps(v) for v in values),False
|
|
return 0,obj['kind']+'/'+obj['metadata']['name'],False
|
|
return 1,'SENTINEL_SECRET',True
|
|
m.invoke=fake
|
|
try:
|
|
result=m.run(payload)
|
|
self.assertTrue(all(x['admitted'] for x in result['dry_runs']))
|
|
deploy=next(x for x in result['dry_runs'] if x['kind']=='Deployment')
|
|
self.assertTrue(deploy['projection_matches']);self.assertEqual(deploy['replicas'],0)
|
|
self.assertFalse(result['held_fence']);self.assertNotIn('SENTINEL_SECRET',json.dumps(result))
|
|
self.assertTrue(all('--dry-run=server' in c for c in calls if c[0]=='kubectl' and c[1]=='apply'))
|
|
finally:m.query=original_query
|
|
if __name__=='__main__':unittest.main()
|