49 lines
3.5 KiB
Python
49 lines
3.5 KiB
Python
import importlib.util,sys,unittest
|
|
from pathlib import Path
|
|
ROOT=Path(__file__).resolve().parents[1];sys.path.insert(0,str(ROOT))
|
|
spec=importlib.util.spec_from_file_location('network_emit',ROOT/'render-c01-network-routes.py')
|
|
emitter=importlib.util.module_from_spec(spec);spec.loader.exec_module(emitter)
|
|
|
|
class EmitTests(unittest.TestCase):
|
|
def inputs(self):
|
|
group='/user.slice/user-1000.slice/user@1000.service/app.slice/frontend.service/runtime'
|
|
def row(pid,cgroup,address,port,inode):
|
|
return {'protocol':'tcp','processes':[{'pid':pid,'cgroup':cgroup,'uid':[1000]*4,
|
|
'socket':{'family':2,'address':address,'port':port,'v6only':None,'net_cookie':'a','inode':inode}}]}
|
|
data={'node':'fixture-node','listeners':[
|
|
row(17,group,'127.0.0.1',7778,1),row(20,'/system.slice/archipelago.service','127.0.0.2',7778,2),
|
|
row(20,'/system.slice/archipelago.service','192.0.2.2',7778,3),row(20,'/system.slice/archipelago.service','127.0.0.1',5678,4)]}
|
|
settings=dict(version=1,node='fixture-node',hostname='fixture-node',boot_id='11111111-1111-4111-8111-111111111111',manager_pid=20,
|
|
runtime=dict(runtime_uid=1000,manager_uid=61001,user_cgroup='/user.slice/user-1000.slice/user@1000.service'),
|
|
namespaces={'host':[4,1],'runtime':[4,2],'managed':[4,3]},
|
|
uplink=dict(executable='/usr/bin/pasta.avx2',sha256='a'*64,wan=['eth0','192.0.2.1','192.0.2.2'],dns=['1.1.1.1'],routing='host-dual-stack',dns_host='100.100.100.100'),
|
|
mdns=None,additional_fragments=[],producers=[dict(source_cgroup=group,source_pid=17,app_id='frontend',target_cgroup=group,
|
|
manifest_sha256='b'*64,producer={'kind':'exact'},required=True)])
|
|
return data,settings
|
|
def test_measured_external_gate_never_becomes_a_raw_publication(self):
|
|
data,settings=self.inputs();receipt,audit=emitter.emit(data,settings)
|
|
self.assertEqual(set(receipt),{'node','boot_id','value'})
|
|
routes=receipt['value']['policy']['routes'];self.assertEqual(len(routes),4)
|
|
gate=next(r for r in routes if r['name'].startswith('gate/'))
|
|
self.assertEqual(gate['source_addresses'],['127.0.0.2','192.0.2.2'])
|
|
self.assertEqual((gate['target'],gate['target_address']),('managed','127.0.0.2'))
|
|
raw=next(r for r in routes if r['name']=='app/frontend/tcp/7778/host')
|
|
self.assertEqual(raw['source_addresses'],['127.0.0.1'])
|
|
self.assertFalse(audit['live_admission'])
|
|
def test_unknown_runtime_producer_refuses_instead_of_dropping_port(self):
|
|
data,settings=self.inputs();settings['producers']=[]
|
|
with self.assertRaisesRegex(ValueError,'producer binding'):emitter.emit(data,settings)
|
|
def test_node_label_cannot_contain_paths_or_change_host_binding(self):
|
|
for label in ('../node','-node','node-','a'*65,True):
|
|
data,settings=self.inputs();settings['node']=label
|
|
with self.subTest(label=label),self.assertRaises(ValueError):emitter.emit(data,settings)
|
|
data,settings=self.inputs();settings['hostname']='another-host'
|
|
with self.assertRaises(ValueError):emitter.emit(data,settings)
|
|
def test_explicit_master_binding_survives_worker_first_socket_inventory(self):
|
|
data,settings=self.inputs();worker={**data['listeners'][0]['processes'][0],'pid':18}
|
|
data['listeners'][0]['processes'].insert(0,worker)
|
|
receipt,_=emitter.emit(data,settings)
|
|
self.assertEqual(len(receipt['value']['policy']['routes']),4)
|
|
|
|
if __name__=='__main__':unittest.main()
|