86 lines
4.8 KiB
Python
86 lines
4.8 KiB
Python
"""Appointment refusal and owned-child lifetime tests; no host commands."""
|
|
import copy
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
|
|
import c01_uplink as uplink
|
|
|
|
|
|
class UplinkTests(unittest.TestCase):
|
|
def test_host_routing_keeps_ndp_and_kernel_source_selection_without_loopback_mapping(self):
|
|
value={'executable':'/usr/bin/pasta.avx2','sha256':'a'*64,
|
|
'wan':['enp1s0','192.0.2.1','192.0.2.2'],'dns':['1.1.1.1'],
|
|
'routing':'host-dual-stack','dns_host':'100.100.100.100'}
|
|
argv=uplink.arguments(value,'/owned-netns')
|
|
for forbidden in ('-4','--no-icmp','--no-ndp','--outbound','--outbound-if4','--outbound-if6','--map-host-loopback'):
|
|
self.assertNotIn(forbidden,argv)
|
|
self.assertIn('--no-map-gw',argv)
|
|
for flag in ('-t','-u','-T','-U'):
|
|
self.assertEqual(argv[argv.index(flag)+1],'none')
|
|
self.assertEqual(argv[argv.index('--dns-host')+1],'100.100.100.100')
|
|
for change in ({'routing':'arbitrary'},{'dns_host':'127.0.0.1'},
|
|
{'dns_host':'192.168.1.1'},{'dns':['8.8.8.8']},
|
|
{'argv':['--map-host-loopback','1.1.1.1']}):
|
|
with self.subTest(change=change),self.assertRaises(ValueError):
|
|
uplink.validate({**value,**change})
|
|
with patch.object(uplink.Path,'read_text',return_value='nameserver 100.100.100.100\n'):
|
|
uplink.check_resolver(value)
|
|
with patch.object(uplink.Path,'read_text',return_value='nameserver 127.0.0.1\n'):
|
|
with self.assertRaisesRegex(ValueError,'no longer configured'):uplink.check_resolver(value)
|
|
|
|
def test_appointment_refuses_argv_host_dns_and_unbound_interface(self):
|
|
value={'executable':'/usr/bin/pasta','sha256':'a'*64,
|
|
'wan':['enp1s0','192.0.2.1','192.0.2.2'],'dns':['1.1.1.1']}
|
|
self.assertEqual(uplink.validate(value),value)
|
|
for change in ({'executable':'/bin/sh'}, {'argv':['--map-host-loopback','1.2.3.4']},
|
|
{'dns':['127.0.0.1']},{'dns':['192.168.1.1']},{'dns':[]},
|
|
{'wan':['x;command','192.0.2.1','192.0.2.2']},
|
|
{'wan':['eth0','127.0.0.1','192.0.2.2']}):
|
|
with self.subTest(change=change),self.assertRaises((ValueError,TypeError)):
|
|
uplink.validate({**copy.deepcopy(value),**change})
|
|
|
|
def test_pidfd_loss_is_terminal_even_if_poll_has_not_reaped(self):
|
|
owner=uplink.Uplink.__new__(uplink.Uplink)
|
|
owner.child=Mock();owner.child.poll.return_value=None;owner.pidfd=10
|
|
with patch.object(uplink.select,'select',return_value=([10],[],[])):
|
|
with self.assertRaisesRegex(ValueError,'egress exited'):owner.alive()
|
|
|
|
def test_second_realm_failure_reaps_first_owned_uplink(self):
|
|
first=Mock()
|
|
with patch.object(uplink,'Uplink',side_effect=[first,ValueError('managed startup failed')]) as factory:
|
|
with self.assertRaisesRegex(ValueError,'managed startup failed'):
|
|
uplink.Uplinks('policy','appointment')
|
|
self.assertEqual([call.args[2] for call in factory.call_args_list],['runtime','managed'])
|
|
first.close.assert_called_once()
|
|
|
|
def test_either_realm_lifetime_loss_fails_combined_owner(self):
|
|
runtime,managed=Mock(),Mock()
|
|
with patch.object(uplink,'Uplink',side_effect=[runtime,managed]):
|
|
owner=uplink.Uplinks('policy','appointment')
|
|
managed.check.side_effect=ValueError('managed egress lost')
|
|
with self.assertRaisesRegex(ValueError,'managed egress lost'):owner.check()
|
|
owner.close()
|
|
runtime.close.assert_called_once();managed.close.assert_called_once()
|
|
|
|
def test_close_reaps_before_namespace_unpin(self):
|
|
owner=uplink.Uplink.__new__(uplink.Uplink)
|
|
owner.child=Mock();owner.child.poll.return_value=None
|
|
owner.pidfd=10;owner.exe=11;owner.mounted=True;owner.target=Path('/owned-ns');owner.created=[]
|
|
order=[];owner.child.wait.side_effect=lambda **kw:order.append('reaped')
|
|
with patch.object(uplink.os,'close'),patch.object(uplink.LIBC,'umount2',side_effect=lambda *args:order.append('unpin') or 0):
|
|
owner.close()
|
|
self.assertEqual(order,['reaped','unpin'])
|
|
owner.child.terminate.assert_called_once()
|
|
|
|
def test_live_pid_with_different_executable_is_not_the_appointed_producer(self):
|
|
owner=uplink.Uplink.__new__(uplink.Uplink)
|
|
owner.child=Mock(pid=321);owner.child.poll.return_value=None
|
|
owner.pidfd=10;owner.exe_identity=(1,2,3,4,5)
|
|
with patch.object(uplink.select,'select',return_value=([],[],[])),patch.object(uplink.os,'stat',return_value=Mock(st_dev=1,st_ino=9)):
|
|
with self.assertRaisesRegex(ValueError,'executable changed'):owner.alive()
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|