import json import io import os import pathlib import sys import tempfile import unittest from unittest.mock import patch sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) import os_update_executor as executor import os_update_plan as plan import os_update_repository as repository def candidate(): return dict(version=1, suite='trixie', architecture='amd64', before_sha256='a'*64, preservation_sha256='b'*64, service_policy='reviewed-maintenance-window', repositories=[dict(pocket=pocket, inrelease_sha256='c'*64, indexes=[dict(path=component + '/binary-amd64/Packages.xz', sha256='d'*64, size=100) for component in ('main', 'contrib', 'non-free', 'non-free-firmware')]) for pocket in ('trixie', 'trixie-updates', 'trixie-security')], packages=[], transaction=[], removals=[], custody=[], boot=dict(running_release='old', expected_release='new', package=dict(name='linux-image-new', version='1', architecture='amd64', sha256='a'*64, size=100, filename='pool/main/l/linux/linux-image-new_1_amd64.deb', pocket='trixie'), retained_files={ '/boot/vmlinuz-old': 'e'*64, '/boot/initrd.img-old': 'f'*64, '/boot/grub/grub.cfg': 'a'*64})) class ContractTests(unittest.TestCase): def test_private_mirror_disables_translations_without_weakening_authentication(self): with tempfile.TemporaryDirectory() as directory: obj = executor.Executor.__new__(executor.Executor) obj.root = pathlib.Path(directory) obj.plan = candidate() with patch.object(executor, 'atomic', lambda path, raw: path.write_bytes(raw)), patch.object( executor.custody, 'secure', lambda path, directory=False: os.open(path, os.O_RDONLY | os.O_NOFOLLOW | (os.O_DIRECTORY if directory else 0))): obj.configure() config = (obj.root / 'apt.conf').read_text() self.assertIn('Acquire::Languages "none";', config) self.assertIn('APT::Get::AllowUnauthenticated "false";', config) self.assertIn('Acquire::Check-Valid-Until "true";', config) self.assertNotIn('trusted=yes', (obj.root / 'sources.list').read_text()) def test_strict_plan_and_no_repository_or_shell_escape(self): self.assertEqual(plan.validate(candidate()), candidate()) for key, value in [('suite', 'stable'), ('architecture', '$(id)'), ('service_policy', 'silent')]: value_plan = candidate() value_plan[key] = value with self.assertRaises(ValueError): plan.validate(value_plan) for path in ('../../etc/apt/apt.conf', 'main/binary-amd64/Packages;id', 'main/binary-arm64/Packages'): value_plan = candidate() value_plan['repositories'][0]['indexes'][0]['path'] = path with self.assertRaises(ValueError): plan.validate(value_plan) value_plan = candidate() value_plan['apt_args'] = ['--allow-unauthenticated'] with self.assertRaises(ValueError): plan.validate(value_plan) def test_removal_requires_exact_disposition_and_custody_refusal(self): value = candidate() value['transaction'] = [dict(operation='Remv', name='obsolete', version='1')] with self.assertRaises(ValueError): plan.validate(value) value['removals'] = ['obsolete'] plan.validate(value) for name in ('podman', 'linux-image-old', 'archipelago', 'cryptsetup'): value['transaction'][0]['name'] = name value['removals'] = [name] with self.assertRaises(ValueError): plan.validate(value) def test_realistic_solver_output_preserves_versions_and_removals(self): result = plan.simulation('Reading package lists...\nInst libc6 [2.36-1] (2.36-2 Debian:12/stable [amd64])\nRemv obsolete [1.0]\nConf libc6 (2.36-2 Debian:12/stable [amd64])\n') self.assertEqual(len(result), 3) self.assertIn(dict(operation='Remv', name='obsolete', version='1.0'), result) with self.assertRaises(ValueError): plan.simulation('Inst unsupported syntax') def test_duplicate_debian_fields_rejected(self): with self.assertRaises(ValueError): repository.paragraphs(b'Package: hello\nSHA256: a\nSHA256: b\n') def test_partial_suite_is_not_a_full_upgrade(self): value = candidate() value['repositories'].pop() with self.assertRaisesRegex(ValueError, 'pocket coverage'): plan.validate(value) value = candidate() value['repositories'][0]['indexes'].pop() with self.assertRaisesRegex(ValueError, 'component coverage'): plan.validate(value) def test_archive_hash_and_size_are_both_enforced(self): for wanted, size in [('a'*64, 3), (plan.digest(b'abc'), 2)]: with patch.object(repository.urllib.request, 'build_opener') as opener: opener.return_value.open.return_value = io.BytesIO(b'abc') with self.assertRaises(ValueError): repository.fetch('https://deb.debian.org/debian/pool/test.deb', wanted, size) def test_bad_debian_signature_stops_before_index_or_archive_download(self): with tempfile.TemporaryDirectory() as directory, patch.object(repository, 'fetch', return_value=b'bad-signature') as fetch, patch.object(repository, 'command', side_effect=ValueError('bad Debian signature')): with self.assertRaisesRegex(ValueError, 'bad Debian signature'): repository.acquire(candidate(), directory) fetch.assert_called_once() def test_no_application_unsigned_mode(self): binding = dict(version=1, plan_sha256='a'*64, manifest_sha256='b'*64, authority_digest='c'*64, verifier_sha256='d'*64, unsigned=True) with self.assertRaises(ValueError): executor.Executor(binding) with self.assertRaises(ValueError): executor.stage({'os_update': {}}, 'RESTORE') def test_release_verifier_failure_blocks_constructor(self): value = candidate() raw = plan.canonical(value) manifest = plan.canonical(dict(os_update=dict(version=1, plan_sha256=plan.digest(raw)))) binding = dict(version=1, plan_sha256=plan.digest(raw), manifest_sha256=plan.digest(manifest), authority_digest='c'*64, verifier_sha256=plan.digest(b'verifier')) def read(path): return {'plan.json': raw, 'manifest.json': manifest, 'archipelago-os-release-verifier': b'verifier'}[path.name] with patch.object(executor.custody, 'Pin'), patch.object(executor, 'read', read), patch.object(executor, 'command', side_effect=ValueError('bad release signature')) as verify: with self.assertRaises(ValueError): executor.Executor(binding) self.assertEqual(verify.call_args.args[0][1:3], ['ceremony', 'verify']) class WorkerTests(unittest.TestCase): def test_preparation_refusal_precedes_manager_effects(self): import d09_worker worker = d09_worker.Worker.__new__(d09_worker.Worker) worker.binding = {'os_update': {}} worker.value = {'receipt': {'phase': 'EXECUTING'}} with patch.object(executor, 'stage', side_effect=ValueError('solver drift')), patch.object(d09_worker.owner, 'command') as command: with self.assertRaises(d09_worker.OSPreparationRefused): worker.execute('key', {'operation': 'APPLY'}) command.assert_not_called() def test_apply_is_reached_from_actual_stopped_worker(self): import d09_worker worker = d09_worker.Worker.__new__(d09_worker.Worker) worker.binding = {'os_update': {}} worker.value = {'receipt': {'phase': 'STOPPED'}} with patch.object(executor, 'stage', side_effect=[None, RuntimeError('apply reached')]) as stage: with self.assertRaisesRegex(RuntimeError, 'apply reached'): worker.execute('key', {'operation': 'APPLY'}) self.assertEqual([call.args[2] for call in stage.call_args_list], ['prepare', 'apply']) class JournalTests(unittest.TestCase): def setUp(self): self.lock_patch = patch.object(executor, 'PackageLease') self.lease = self.lock_patch.start().return_value.__enter__.return_value self.lease.observation = [] self.addCleanup(self.lock_patch.stop) def make_executor(self, directory): obj = executor.Executor.__new__(executor.Executor) obj.root = pathlib.Path(directory) obj.status_path = obj.root / 'status.json' obj.status = dict(version=1, plan_sha256='a'*64, authority_digest='b'*64, phase='VERIFIED', sequence=1, events=[]) obj.plan = candidate() return obj def test_partial_dpkg_failure_is_durable_and_never_retried(self): with tempfile.TemporaryDirectory() as directory: obj = self.make_executor(directory) writes = [] def write(path, raw): pathlib.Path(path).write_bytes(raw) writes.append(json.loads(raw)['phase']) self.lease.install.side_effect = RuntimeError('postinst failed') with patch.object(executor, 'atomic', write), patch.object(obj, 'preflight'), patch.object(obj, 'solver'), patch.object(obj, 'verify_archives'), patch.object(obj, 'apt', return_value='') as apt: with self.assertRaises(RuntimeError): obj.apply() self.assertEqual(writes, ['APPLY_INTENT', 'RECONCILIATION_REQUIRED']) obj.status = json.loads(obj.status_path.read_bytes()) with self.assertRaises(executor.ReconciliationRequired): obj.apply() apt.assert_called_once_with('update') self.lease.install.assert_called_once() def test_crash_intent_cannot_resume_or_restore(self): with tempfile.TemporaryDirectory() as directory: obj = self.make_executor(directory) for phase in ('APPLY_INTENT', 'VERIFY', 'RECONCILIATION_REQUIRED', 'DOWNLOAD'): obj.status['phase'] = phase with patch.object(obj, 'apt') as apt: with self.assertRaises(executor.ReconciliationRequired): obj.apply() apt.assert_not_called() def test_lost_completion_reply_does_not_rewrite_done_journal(self): with tempfile.TemporaryDirectory() as directory: obj = self.make_executor(directory) def checkpoint(): if obj.status['phase'] == 'REBOOT_REQUIRED': raise ConnectionError('lost owner reply after completion publication') obj.checkpoint = checkpoint with patch.object(executor, 'atomic', lambda path, raw: path.write_bytes(raw)), patch.object(obj, 'preflight'), patch.object(obj, 'solver'), patch.object(obj, 'verify_archives'), patch.object(obj, 'apt'), patch.object(obj, 'verify_installed'), patch.object(obj, 'boot_provenance', return_value={}), patch.object(executor, 'inventory', return_value=b'fixture'): with self.assertRaises(ConnectionError): obj.apply() persisted = obj.status_path.read_bytes() self.assertEqual(json.loads(persisted)['phase'], 'REBOOT_REQUIRED') obj.apply() self.assertEqual(obj.status_path.read_bytes(), persisted) self.lease.install.assert_called_once() def test_solver_mismatch_precedes_consumption_and_effects(self): with tempfile.TemporaryDirectory() as directory: obj = self.make_executor(directory) with patch.object(obj, 'preflight'), patch.object(obj, 'verify_archives'), patch.object(obj, 'solver', side_effect=ValueError('drift')), patch.object(obj, 'apt') as apt: with self.assertRaises(ValueError): obj.apply() self.assertEqual(obj.status['phase'], 'VERIFIED') apt.assert_called_once_with('update') def test_postboot_requires_new_boot_and_preservation(self): with tempfile.TemporaryDirectory() as directory: obj = self.make_executor(directory) obj.status['phase'] = 'REBOOT_REQUIRED' obj.status['events'] = [dict(phase='REBOOT_REQUIRED', details=dict(boot_id='old', boot=dict(release='new', package_version='1', files={}))) ] with patch.object(executor, 'boot_id', return_value='old'): with self.assertRaises(ValueError): obj.confirm_postboot() obj.source = pathlib.Path(directory) with patch.object(executor, 'boot_id', return_value='new'), patch.object(executor.platform, 'release', return_value='new'), patch.object(executor.platform, 'version', return_value='#1 Debian 1 test'): with self.assertRaises((ValueError, PermissionError, FileNotFoundError)): obj.confirm_postboot() if __name__ == '__main__': unittest.main()