"""Composed protected owner/store/client fixtures; never host services/packages.""" import copy import importlib.util import json import os import pathlib import subprocess import sys import tempfile import types import unittest from unittest.mock import patch sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) import d09_wire as wire import os_update_executor as engine import os_update_owner as os_owner import os_update_lock as locks from test_os_update import candidate path = pathlib.Path(__file__).resolve().parents[2] / 'tests/regression/d09-lifecycle.py' spec = importlib.util.spec_from_file_location('os_d09_fixture', path) lifecycle = importlib.util.module_from_spec(spec) spec.loader.exec_module(lifecycle) class OwnerComposition(unittest.TestCase): def setUp(self): self.fixture = lifecycle.Lifecycle() self.fixture.setUp() self.addCleanup(self.fixture.tearDown) temporary = tempfile.TemporaryDirectory(prefix='os-owner-') self.addCleanup(temporary.cleanup) self.root = pathlib.Path(temporary.name) / 'os' self.root.mkdir(mode=0o700) (self.root / 'plans').mkdir(mode=0o700) (self.root / 'transactions').mkdir(mode=0o700) code = self.root / 'code' code.mkdir(mode=0o700) for name in os_owner.MODULES: (code / name).write_bytes(pathlib.Path(os_owner.__file__).with_name(name).read_bytes()) (code / name).chmod(0o600) self.binding = self.fixture.binding preservation = wire.canonical(dict(authority_digest='9'*64, service_policy='reviewed-maintenance-window', preserve_wallets_data_identity_stop_uninstall=True)) plan = candidate() plan['preservation_sha256'] = wire.digest(preservation) raw = wire.canonical(plan) identity = ['B', [['archipelago', 'B', 'e'*64, 1], ['os-update-plan.json', 'B', wire.digest(raw), len(raw)]]] manifest = wire.canonical(dict(version='B', os_update=dict(version=1, plan_sha256=wire.digest(raw)), components=[dict(zip(('name','new_version','sha256','size_bytes'), row)) for row in identity[1]])) verifier = self.root / 'verifier' verifier.write_bytes(b'\x7fELF disposable externally appointed verifier') verifier.chmod(0o700) self.binding['release_identity'] = identity self.binding['os_update'] = dict(version=1, plan_sha256=wire.digest(raw), manifest_sha256=wire.digest(manifest), verifier_sha256=wire.digest(verifier.read_bytes()), authority_digest='9'*64) self.binding['prerequisite']['backend_sha256'] = wire.digest(verifier.read_bytes()) lifecycle.store.atomic(self.root / 'verifier-custody.json', wire.canonical(dict(version=1, verifier_sha256=wire.digest(verifier.read_bytes()), prerequisite_backend_sha256=wire.digest(verifier.read_bytes())))) source = self.root / 'plans' / wire.digest(raw) source.mkdir(mode=0o700) for name, data in [('plan.json', raw), ('manifest.json', manifest), ('preservation.json', preservation)]: (source / name).write_bytes(data) (source / name).chmod(0o600) self.patches = [patch.object(engine, 'ROOT', self.root), patch.object(os_owner, 'ROOT', self.root), patch.object(engine.custody, 'secure', lifecycle.fixture_secure), patch.object(engine.custody, 'LEGACY_ROOT', self.root / 'unused-parent/os-update'), patch.object(os_owner, '__file__', str(code / 'os_update_owner.py')), patch.object(engine, 'VERIFIER', verifier), patch.object(os_owner, 'VERIFIER', verifier), patch.object(engine, 'secure', lifecycle.fixture_secure), patch.object(os_owner, 'secure', lifecycle.fixture_secure), patch.object(engine, 'command', return_value=''), patch.object(os_owner, 'command', side_effect=self.query)] for p in self.patches: p.start() self.addCleanup(p.stop) def query(self, args): if args[-1] == 'base-files': return '13.2\tii ' raise subprocess.CalledProcessError(1, args) # new kernel not installed def inspect(self): req = dict(schema_version=1, target_binding_digest=self.binding['target_binding_digest'], transaction_id=None, challenge='8'*64) response = self.fixture.o.inspect(req, self.fixture.peer) return wire.bind_inspect(req, wire.decode('inspect_response', wire.canonical(response))) def test_qualified_owner_projection_reaches_existing_client_without_new_kernel(self): result = self.inspect() status = result['os_update'] self.assertTrue(status['supported']) self.assertEqual(status['phase'], 'NO_OS_TRANSACTION') self.assertEqual(status['plan_sha256'], self.binding['os_update']['plan_sha256']) self.assertEqual(status['installed_os'], 'Debian trixie / base-files 13.2') self.assertIsNone(status['installed_kernel']) spec = importlib.util.spec_from_file_location('composed_os_client', pathlib.Path(wire.__file__).with_name('d09-lifecycle-client.py')) client = importlib.util.module_from_spec(spec) spec.loader.exec_module(client) cache = self.root / 'client' cache.mkdir(mode=0o700) with patch.object(client, 'CACHE', cache), patch.object(client, 'inspect', side_effect=lambda _: self.inspect()), patch.object(client, 'exchange', side_effect=lambda req, _: self.fixture.submit(req)): accepted = client.submit_cached(self.binding, 'prepare', 'UPDATER', '5'*32, 'fixture', cache / 'request.json', 'a'*32) self.assertEqual(accepted['status'], 'ACCEPTED') def test_signed_manifest_component_mismatch_refuses_capability(self): self.binding['release_identity'][1][0][2] = '7'*64 self.assertFalse(self.inspect()['os_update']['supported']) def test_actual_provisioner_admits_os_inputs_and_bootstraps_owner(self): spec = importlib.util.spec_from_file_location('composed_os_provisioner', pathlib.Path(wire.__file__).with_name('provision-d09-lifecycle.py')) provisioner = importlib.util.module_from_spec(spec) spec.loader.exec_module(provisioner) root = self.root / 'provisioned' root.mkdir(mode=0o700) for name in ('incoming', 'incoming/os-update', 'run', 'config'): (root / name).mkdir(mode=0o700) source = self.root / 'plans' / self.binding['os_update']['plan_sha256'] incoming = root / 'incoming/os-update' for name in ('plan.json', 'manifest.json', 'preservation.json'): lifecycle.store.atomic(incoming / name, (source / name).read_bytes()) lifecycle.store.atomic(incoming / 'verifier', engine.VERIFIER.read_bytes()) lifecycle.store.atomic(root / 'incoming/exclusion.json', b'external fixture exclusion') binding = {k:v for k,v in self.binding.items() if k not in ('requester_uid', 'requester_gid')} binding.update(objects={}, artifacts={'backend':'e'*64}, restore_artifacts={'backend':'f'*64}, effective_config={'APPLY':'e'*64, 'RESTORE':'f'*64}, exclusion_evidence_digest=wire.digest(b'external fixture exclusion'), helper_digest=wire.digest(b'fixture helper')) admission = root / 'admission.json' lifecycle.store.atomic(admission, wire.canonical(binding)) paths = tuple(self.root / n for n in ('probe-frontend', 'probe-inner')) for path in paths: path.touch(mode=0o600) account = types.SimpleNamespace(pw_dir='/home/archipelago', pw_uid=os.getuid(), pw_gid=os.getgid()) real_read = provisioner.owner.read def helper_read(path): if str(path) == '/usr/libexec/archipelago-d09-prepare-workspace': return b'fixture helper' return real_read(path) with patch.object(provisioner.owner, 'ROOT', root), patch.object(provisioner.owner, 'RUN', root / 'run'), patch.object(provisioner.owner, 'PERMIT', root / 'run/permit'), patch.object(provisioner.owner, 'BINDING', root / 'config/binding.json'), patch.object(provisioner.owner, 'destinations', return_value={}), patch.object(provisioner.owner, 'service', return_value={'MainPID':'0', 'ActiveState':'inactive'}), patch.object(provisioner.owner, 'read', side_effect=helper_read), patch.object(provisioner.pwd, 'getpwnam', return_value=account), patch.object(engine.Executor, 'preflight') as preflight, patch.object(os_owner, 'probe', side_effect=lambda: locks.probe(paths)), patch.object(locks, 'secure', lifecycle.fixture_secure): with patch.object(provisioner, 'read', side_effect=helper_read): provisioner.provision(admission) preflight.assert_called_once() new_owner = provisioner.owner.Owner() try: new_owner.authorize = lambda req, peer: None req = dict(schema_version=1, target_binding_digest=binding['target_binding_digest'], transaction_id=None, challenge='8'*64) result = wire.bind_inspect(req, new_owner.inspect(req, self.fixture.peer)) self.assertTrue(result['os_update']['supported']) self.assertEqual(new_owner.store.value['os_admission']['plan_sha256'], binding['os_update']['plan_sha256']) finally: os.close(new_owner.store.fd) os.close(new_owner.store.head_fd) def test_changed_verifier_refuses_capability(self): engine.VERIFIER.write_bytes(b'\x7fELF different') self.assertFalse(self.inspect()['os_update']['supported']) def test_owner_postboot_observation_is_root_only_and_durable(self): obj = engine.Executor(self.binding['os_update']) for phase in ('PREPARE', 'DOWNLOAD', 'VERIFIED', 'APPLY_INTENT', 'VERIFY', 'REBOOT_REQUIRED'): obj.journal(phase) self.fixture.o.store.checkpoint_os(obj.status) obj.journal('POSTBOOT_CONFIRMED') observation = dict(plan_sha256=self.binding['os_update']['plan_sha256'], boot_id='b'*32, status_sha256=wire.digest(obj.status_path.read_bytes())) with patch.object(os_owner, 'postboot', return_value=observation): with self.assertRaises(PermissionError): self.fixture.o.worker_call({'method':'os_postboot'}, (123, 1000, 1000)) reply = self.fixture.o.worker_call({'method':'os_postboot'}, (123, 0, 0)) self.assertTrue(reply['os_postboot_verified']) self.assertEqual(self.fixture.o.store.value['os_observations'], [observation]) changed = copy.deepcopy(self.fixture.o.store.value) changed['os_observations'] = [] with self.assertRaises(lifecycle.store.ContinuityError): self.fixture.o.store.commit_head(changed) def test_checkpoint_replay_and_unanchored_journal_are_refused(self): obj = engine.Executor(self.binding['os_update']) obj.journal('PREPARE') self.assertEqual(self.inspect()['os_update']['phase'], 'UNKNOWN') self.fixture.o.store.checkpoint_os(obj.status) self.assertEqual(self.inspect()['os_update']['phase'], 'PREPARE') with self.assertRaises(lifecycle.store.ContinuityError): self.fixture.o.store.checkpoint_os(obj.status) obj.journal('DOWNLOAD') self.fixture.o.store.checkpoint_os(obj.status) earlier = copy.deepcopy(obj.status) earlier['events'].pop() earlier.update(sequence=1, phase='PREPARE') lifecycle.store.atomic(obj.status_path, wire.canonical(earlier)) self.assertEqual(self.inspect()['os_update']['phase'], 'UNKNOWN') def test_actual_worker_checkpoint_hook_arms_and_releases_only_existing_stage(self): import os_update_manager obj = engine.Executor(self.binding['os_update']) for phase in ('PREPARE', 'DOWNLOAD', 'VERIFIED'): obj.journal(phase) self.fixture.o.store.checkpoint_os(obj.status) value = copy.deepcopy(self.fixture.o.store.value) unit = 'archipelago-d09-' + 'a'*32 + '-' + 'c'*32 + '.service' value['execution'] = dict(unit=unit, incarnation_id='c'*32, executor=None) self.fixture.o.store.commit_head(value) grants = list(value['grants']) self.fixture.o.incarnation = '9'*32 actual = dict(pid=os.getpid(), start_ticks=1, boot_id=self.fixture.o.boot) original_read = pathlib.Path.read_text def read_text(path, *args, **kwargs): if str(path) == '/proc/%d/cgroup' % os.getpid(): return '0::/system.slice/' + unit + '\n' return original_read(path, *args, **kwargs) policy = dict(operation='debian-systemd-daemon-reexec') # Kernel/bus proofs have their own real PID1 fixture. This test exercises # the production private-handler -> checkpoint/store integration itself. with patch.object(os_update_manager, 'Window') as window, patch.object( os_owner, 'qualify', return_value=(types.SimpleNamespace(plan={'manager_reexec':policy}), None)), patch.object( lifecycle.owner, 'process', return_value=actual), patch.object(pathlib.Path, 'read_text', read_text): window.return_value.state = 'ARMED' window.return_value.deadline = None window.return_value.poll.return_value = True for phase in ('APPLY_INTENT', 'VERIFY', 'REBOOT_REQUIRED'): obj.journal(phase) self.fixture.o.worker_call(dict(method='os_checkpoint', incarnation_id='c'*32), (os.getpid(), 0, 0)) window.assert_called_once_with(self.fixture.o, os.getpid(), policy) window.return_value.close.assert_called_once() self.assertIsNone(self.fixture.o.manager_window) self.assertEqual(self.fixture.o.store.value['grants'], grants) self.assertEqual(self.fixture.o.store.value['os_checkpoints'][-1]['phase'], 'REBOOT_REQUIRED') def test_later_boot_does_not_replay_confirmed_status(self): obj, _ = os_owner.qualify(self.binding) obj.status['phase'] = 'POSTBOOT_CONFIRMED' with patch.object(os_owner, 'qualify', return_value=(obj, None)), patch.object(obj, 'observe_postboot', side_effect=ValueError('another boot')): status = os_owner.projection(self.binding) self.assertEqual(status['phase'], 'REBOOT_REQUIRED') self.assertFalse(status['post_boot_verified']) def test_actual_package_lock_probe_refuses_another_process(self): paths = tuple(self.root / n for n in ('lock-frontend', 'lock')) for path in paths: path.touch(mode=0o600) ready_r, ready_w = os.pipe() release_r, release_w = os.pipe() child = os.fork() if child == 0: try: import fcntl fd = os.open(paths[0], os.O_RDWR) fcntl.lockf(fd, fcntl.LOCK_EX) os.write(ready_w, b'1') os.read(release_r, 1) os._exit(0) except BaseException: os._exit(1) try: os.read(ready_r, 1) with patch.object(locks, 'secure', lifecycle.fixture_secure): with self.assertRaises(BlockingIOError), locks.probe(paths): pass os.write(release_w, b'1') os.waitpid(child, 0) child = None with patch.object(locks, 'secure', lifecycle.fixture_secure), locks.probe(paths) as evidence: self.assertEqual({e['pid'] for e in evidence}, {os.getpid()}) self.assertEqual(len(evidence), 2) finally: if child is not None: os.write(release_w, b'1') os.waitpid(child, 0) for fd in (ready_r, ready_w, release_r, release_w): os.close(fd) if __name__ == '__main__': unittest.main()