277 lines
16 KiB
Python
277 lines
16 KiB
Python
"""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()
|