97 lines
5.4 KiB
Python
97 lines
5.4 KiB
Python
"""Source contracts only. Never creates accounts, namespaces, grants or units."""
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from unittest.mock import patch, Mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import c01_network as network
|
|
import c01_supervisor as supervisor
|
|
import c01_access_kernel as kernel
|
|
|
|
|
|
class NetworkTests(unittest.TestCase):
|
|
def test_kernel_type_refuses_regular_file_and_mount_namespace(self):
|
|
for path in ('/dev/null', '/proc/self/ns/mnt'):
|
|
fd = os.open(path, os.O_RDONLY)
|
|
try:
|
|
with self.assertRaises((ValueError, OSError)):
|
|
network.identity(fd)
|
|
finally:
|
|
os.close(fd)
|
|
fd = os.open('/proc/self/ns/net', os.O_RDONLY)
|
|
try:
|
|
self.assertEqual(network.identity(fd), network.process_namespace(os.getpid()))
|
|
finally:
|
|
os.close(fd)
|
|
|
|
def test_host_or_unappointed_private_namespace_refuses(self):
|
|
with patch.object(network, 'appointed', return_value=(1, 2)):
|
|
with patch.object(network, 'process_namespace', return_value=(1, 2)), self.assertRaisesRegex(ValueError, 'host'):
|
|
network.require_private(1234, 'managed')
|
|
with patch.object(network, 'process_namespace', side_effect=[(1, 3), (1, 4)]), self.assertRaisesRegex(ValueError, 'appointed'):
|
|
network.require_private(1234, 'managed')
|
|
with patch.object(network, 'process_namespace', side_effect=[(1, 3), (1, 2)]):
|
|
self.assertEqual(network.require_private(1234, 'managed'), (1, 2))
|
|
|
|
def test_supervisor_exact_group_nnp_and_namespace(self):
|
|
policy = types.SimpleNamespace(value={'supervisor_cgroup':'/sys/fs/cgroup/system.slice/archipelago.service'})
|
|
item = dict(pid=1234, start=7, uids=[0]*4, gids=[0]*4, nnp=1,
|
|
group='/sys/fs/cgroup/system.slice/archipelago.service/supervisor')
|
|
task=Mock(start=7);task.record.return_value=item;task.finish.return_value=True
|
|
task.namespace.return_value=(1,2);task.dependencies=[]
|
|
with patch.object(kernel,'PinnedTask') as pin, patch.object(kernel,'network_appointment',return_value=(1,2)):
|
|
pin.return_value.__enter__.return_value=task
|
|
self.assertEqual(kernel.verify_supervisor(item, policy), (1, 2))
|
|
pin.assert_called_once_with(1234);task.namespace.assert_called_once_with('net')
|
|
for altered in (dict(item, nnp=0), dict(item, group=item['group']+'/other'), dict(item, uids=[1000]*4)):
|
|
with self.assertRaises(ValueError):
|
|
kernel.verify_supervisor(altered, policy)
|
|
task.finish.return_value=False
|
|
with self.assertRaisesRegex(ValueError,'mandatory supervisor exited'):
|
|
kernel.verify_supervisor(item,policy)
|
|
|
|
|
|
class AdmissionTests(unittest.TestCase):
|
|
def test_persisted_flags_and_duck_typed_leases_cannot_admit(self):
|
|
for owner in (None, types.SimpleNamespace(access_lease=Mock(), runtime_lease=Mock())):
|
|
with self.assertRaises(supervisor.AdmissionUnavailable):
|
|
supervisor.admit({'migration_complete':True,'old_writers_excluded':True}, None, None, owner)
|
|
|
|
def test_partial_acquisition_remains_owned_on_runtime_failure(self):
|
|
owner = types.SimpleNamespace(access_lease=None, runtime_lease=None, network_lease=None)
|
|
value = dict(runtime_uid=1000, runtime_gid=1000, runtime_data_dir='/runtime',
|
|
manager_backing='/private-manager/data', snapshot_backing='/private-snapshot/data',
|
|
authority_backing='/private-authority/data')
|
|
app = types.SimpleNamespace(pw_uid=61001, pw_gid=61001)
|
|
authority = types.SimpleNamespace(pw_uid=61002, pw_gid=61002)
|
|
runtime = dict(manager_uid=61001,manager_gid=61001,authority_uid=61002,authority_gid=61002,
|
|
runtime_uid=1000,runtime_gid=1000,runtime_data_dir='/runtime',home='/home/runtime',
|
|
private_parents=['/private-manager','/private-snapshot','/private-authority'])
|
|
access = Mock()
|
|
with patch.object(network, 'require_private'), patch.object(supervisor, 'root_json', return_value=(value,'a'*64)), \
|
|
patch.object(supervisor.c01_access_client, 'AccessLease', return_value=access), \
|
|
patch.object(supervisor.runtime_service, 'load_config', return_value=(runtime,'b'*64)), \
|
|
patch.object(supervisor.c01_runtime_view, 'RuntimeLease', side_effect=ValueError('runtime unavailable')):
|
|
with self.assertRaisesRegex(ValueError, 'runtime unavailable'):
|
|
supervisor.acquire_admission_leases(owner, value, app, authority)
|
|
self.assertIs(owner.access_lease, access)
|
|
self.assertIsNone(owner.runtime_lease)
|
|
access.close.assert_not_called() # retained until Supervisor.retire
|
|
|
|
def test_shared_access_directory_has_no_supervisor_cleanup_owner(self):
|
|
path = Path(__file__).resolve().parents[2] / 'image-recipe/configs/70-c01-supervisor.conf.in'
|
|
text = path.read_text()
|
|
values = [line.split('=',1)[1] for line in text.splitlines() if line.startswith('RuntimeDirectory=')]
|
|
self.assertEqual(values, [''])
|
|
self.assertIn('EnvironmentFile=\n', text)
|
|
self.assertIn('ExecStartPre=\n', text)
|
|
self.assertIn('NetworkNamespacePath=/run/netns/archipelago-managed', text)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|