78 lines
4.9 KiB
Python
78 lines
4.9 KiB
Python
import copy,hashlib,json,socket,sys,unittest
|
|
from pathlib import Path
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
|
|
import c01_network_plan as plan
|
|
import c01_network_owner as owner
|
|
import c01_network_mdns as mdns
|
|
|
|
|
|
class V4Tests(unittest.TestCase):
|
|
def test_conditional_admission_covers_the_full_128_route_policy_bound(self):
|
|
import c01_admission as admission
|
|
value=dict(version=1,platform_sha256='a'*64,roles=sorted(admission.ROLES),
|
|
assets={name:'b'*64 for name in admission.ASSETS},services={name:'uninstalled' for name in admission.UNITS},
|
|
optional_unavailable=sorted(admission.UNITS),required_routes=['management']+[f'route-{n}' for n in range(127)])
|
|
admission.validate(value,{'platform_profile_sha256':'a'*64})
|
|
value['required_routes'].append('one-too-many')
|
|
with self.assertRaisesRegex(ValueError,'required route'):admission.validate(value,{'platform_profile_sha256':'a'*64})
|
|
|
|
def test_wildcard_conflicts_are_family_specific(self):
|
|
from c01_endpoints import source_binding_conflict
|
|
key=lambda address:('host','udp',5353,address)
|
|
self.assertFalse(source_binding_conflict(key('0.0.0.0'),key('::')))
|
|
self.assertTrue(source_binding_conflict(key('0.0.0.0'),key('127.0.0.1')))
|
|
self.assertTrue(source_binding_conflict(key('::'),key('fd00::1')))
|
|
|
|
def fixture(self):
|
|
runtime={'runtime_uid':1000,'runtime_gid':1000,'manager_uid':61001,
|
|
'user_cgroup':'/user.slice/user-1000.slice/user@1000.service'}
|
|
group=runtime['user_cgroup']+'/app.slice/fixture.service'
|
|
doc={'app_id':'fixture','manifest_sha256':'a'*64,'loopback_ports':[['tcp',7778]],'target_cgroup':group,
|
|
'source_bindings':[['host','tcp',7778,'0.0.0.0'],['host','tcp',7778,'::']],
|
|
'target_addresses':['127.0.0.1'],'producer':{'kind':'exact'}}
|
|
digest=hashlib.sha256(json.dumps(doc,sort_keys=True,separators=(',',':')).encode()).hexdigest()
|
|
route=dict(name='app/fixture/tcp/7778/host',source='host',target='runtime',protocol='tcp',port=7778,
|
|
target_port=7778,app_id='fixture',projection_sha256=digest,target_cgroup=group,target_uid=1000,
|
|
source_addresses=['0.0.0.0','::'],target_address='127.0.0.1')
|
|
policy={'version':4,'namespaces':{'host':[4,1],'managed':[4,2],'runtime':[4,3]},'manifests':[doc],'routes':[route],
|
|
'uplink':{'executable':'/usr/bin/pasta.avx2','sha256':'b'*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'}}
|
|
return policy,runtime
|
|
|
|
def test_exact_dual_stack_publication_and_source_projection(self):
|
|
policy,runtime=self.fixture()
|
|
self.assertEqual(plan.validate(policy,runtime)['listeners'],2)
|
|
policy['routes'][0]['source_addresses']=['192.0.2.9']
|
|
with self.assertRaisesRegex(ValueError,'qualified address'):plan.validate(policy,runtime)
|
|
|
|
def test_public_runtime_alias_cannot_bypass_gate(self):
|
|
policy,runtime=self.fixture()
|
|
doc={'app_id':'fixture','manifest_sha256':'c'*64,'loopback_ports':[['tcp',7778]],
|
|
'target_cgroup':'/system.slice/archipelago.service/c01-application'}
|
|
digest=hashlib.sha256(json.dumps(doc,sort_keys=True,separators=(',',':')).encode()).hexdigest()
|
|
policy['manifests'].append(doc)
|
|
policy['routes'].append(dict(name='gate/fixture/tcp/17778/host',source='host',target='managed',protocol='tcp',
|
|
port=17778,target_port=7778,app_id='fixture',projection_sha256=digest,target_cgroup=doc['target_cgroup'],
|
|
target_uid=61001,source_addresses=['192.0.2.2'],target_address='127.0.0.2'))
|
|
with self.assertRaisesRegex(ValueError,'bypass'):plan.validate(policy,runtime)
|
|
|
|
def test_mdns_scope_cannot_select_other_interfaces_or_broad_operations(self):
|
|
policy,_=self.fixture();scope={'interface':'eth0','ifindex':2,'ipv4':'192.0.2.2'}
|
|
mdns.validate(scope,policy['uplink'])
|
|
for change in ({'interface':'lo'},{'ipv4':'127.0.0.1'},{'ifindex':0},{'argv':['ip','route','add']}):
|
|
with self.subTest(change=change),self.assertRaises(ValueError):mdns.validate({**scope,**change},policy['uplink'])
|
|
|
|
def test_gate_prefix_uses_kernel_peer_and_has_bounded_v2_shape(self):
|
|
with socket.socket() as listener:
|
|
listener.bind(('127.0.0.1',0));listener.listen(1)
|
|
with socket.create_connection(listener.getsockname()) as client:
|
|
connection,_=listener.accept()
|
|
with connection:
|
|
header=owner.gate_peer_header(connection)
|
|
self.assertEqual(header[:16],b'\r\n\r\n\0\r\nQUIT\n\x21\x11\x00\x0c')
|
|
self.assertEqual(header[16:20],socket.inet_aton(client.getsockname()[0]))
|
|
self.assertEqual(len(header),28)
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|