240 lines
13 KiB
Python
240 lines
13 KiB
Python
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()
|