78 lines
4.6 KiB
Python
78 lines
4.6 KiB
Python
|
import copy
|
||
|
|
import importlib.util
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
spec=importlib.util.spec_from_file_location('render_deployment',Path(__file__).parents[1]/'tools/render_deployment.py')
|
||
|
|
module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
|
||
|
|
|
||
|
|
|
||
|
|
def inputs():
|
||
|
|
return {'image':'forgejo.coulomb.social/coulomb/informed-decision@sha256:'+'a'*64,
|
||
|
|
'policy':{'origin':'http://flex-auth-review.flex-auth.svc.cluster.local:8080','package':'review.example',
|
||
|
|
'version':'v1','package_digest':'sha256:'+'b'*64,'pod_name':'flex-auth-review'},
|
||
|
|
'keycape_egress_ips':['92.205.62.239'],'storage_class':'local-path'}
|
||
|
|
|
||
|
|
|
||
|
|
def test_candidate_does_not_create_custody_and_uses_one_private_writer():
|
||
|
|
result=module.render(inputs());objects=result['items']
|
||
|
|
assert not any(o['kind'] in {'Secret','Role','RoleBinding','ClusterRoleBinding','Ingress','Namespace'} for o in objects)
|
||
|
|
deploy=next(o for o in objects if o['kind']=='Deployment')['spec']
|
||
|
|
assert deploy['replicas']==1 and deploy['strategy']=={'type':'Recreate'}
|
||
|
|
pod=deploy['template']['spec'];container=pod['containers'][0]
|
||
|
|
assert pod['automountServiceAccountToken'] is False and pod['serviceAccountName']=='review'
|
||
|
|
assert pod['securityContext']['runAsUser']==pod['securityContext']['fsGroup']==10001
|
||
|
|
assert container['securityContext']['readOnlyRootFilesystem'] is True
|
||
|
|
assert container['securityContext']['capabilities']['drop']==['ALL']
|
||
|
|
caller=next(v for v in pod['volumes'] if v['name']=='caller')
|
||
|
|
assert caller['projected']['sources']==[{'serviceAccountToken':{'path':'token','audience':'flex-auth','expirationSeconds':3600}}]
|
||
|
|
# Healthy serving remains available for refusal/recovery; application itself
|
||
|
|
# gates accept on audit /readyz, without a dependency-driven restart cycle.
|
||
|
|
assert container['readinessProbe']['httpGet']['path']=='/healthz'
|
||
|
|
assert container['livenessProbe']['httpGet']['path']=='/healthz'
|
||
|
|
|
||
|
|
|
||
|
|
def test_policies_and_service_selectors_do_not_include_placeholder_or_whole_namespaces():
|
||
|
|
objects=module.render(inputs())['items']
|
||
|
|
service=next(o for o in objects if o['kind']=='Service')
|
||
|
|
assert service['spec']['selector']['app.kubernetes.io/component']=='review'
|
||
|
|
policy=next(o for o in objects if o['kind']=='NetworkPolicy')['spec']
|
||
|
|
for rule in policy['ingress']+policy['egress']:
|
||
|
|
for peer in rule.get('from',rule.get('to',[])):
|
||
|
|
if 'namespaceSelector' in peer:assert 'podSelector' in peer
|
||
|
|
assert policy['egress'][-1]['to']==[{'ipBlock':{'cidr':'92.205.62.239/32'}}]
|
||
|
|
assert policy['egress'][-1]['ports']==[{'protocol':'TCP','port':443}]
|
||
|
|
assert policy['egress'][-2]['to'][0]['podSelector']['matchLabels']=={'app.kubernetes.io/name':'traefik'}
|
||
|
|
assert policy['egress'][-2]['ports']==[{'protocol':'TCP','port':8443}]
|
||
|
|
peers=[o for o in objects if o['kind']=='NetworkPolicy' and o['metadata']['namespace']!='informed-decision']
|
||
|
|
assert {o['metadata']['namespace'] for o in peers}=={'approval-engine','flex-auth'}
|
||
|
|
for obj in peers:
|
||
|
|
caller=obj['spec']['ingress'][0]['from'][0]
|
||
|
|
assert caller['namespaceSelector']['matchLabels']=={'kubernetes.io/metadata.name':'informed-decision'}
|
||
|
|
assert caller['podSelector']['matchLabels']['app.kubernetes.io/component']=='review'
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('fault',['tag','other_image','missing_pin','external_pdp','wildcard_pod','unknown_field',
|
||
|
|
'broad_egress','loopback','empty_ips','empty_storage','empty_package'])
|
||
|
|
def test_incomplete_or_broad_deployment_inputs_refuse(fault):
|
||
|
|
data=inputs()
|
||
|
|
if fault=='tag':data['image']='forgejo.coulomb.social/coulomb/informed-decision:latest'
|
||
|
|
if fault=='other_image':data['image']=data['image'].replace('informed-decision','other')
|
||
|
|
if fault=='missing_pin':del data['policy']['package_digest']
|
||
|
|
if fault=='external_pdp':data['policy']['origin']='https://elsewhere.test'
|
||
|
|
if fault=='wildcard_pod':data['policy']['pod_name']='*'
|
||
|
|
if fault=='unknown_field':data['token']='not-permitted-inline'
|
||
|
|
if fault=='broad_egress':data['keycape_egress_ips']=['0.0.0.0/0']
|
||
|
|
if fault=='loopback':data['keycape_egress_ips']=['127.0.0.1']
|
||
|
|
if fault=='empty_ips':data['keycape_egress_ips']=[]
|
||
|
|
if fault=='empty_storage':data['storage_class']=''
|
||
|
|
if fault=='empty_package':data['policy']['package']=''
|
||
|
|
with pytest.raises(ValueError):module.render(data)
|
||
|
|
|
||
|
|
|
||
|
|
def test_policy_revision_changes_immutable_configuration_name():
|
||
|
|
first=inputs();second=copy.deepcopy(first);second['policy']['version']='v2'
|
||
|
|
names=lambda data:[o['metadata']['name'] for o in module.render(data)['items'] if o['kind']=='ConfigMap']
|
||
|
|
assert names(first)!=names(second)
|