"""Actual owner/journal/wire/client integration; no host package or service effects. Only OS observations, fixture custody and the appointed verifier process are adapted. The verifier adapter performs real Ed25519 verification using a disposable fixture key; production release-root verification is separately tested in Rust. Owner.inspect, qualify, Executor loading, wire binding and client projection are the production implementations, not synthesized status DTOs. """ import importlib.util import json from pathlib import Path import subprocess import sys import types import unittest from unittest.mock import patch SCRIPTS = Path(__file__).resolve().parents[1] sys.path.insert(0, str(SCRIPTS)) sys.path.insert(0, str(Path(__file__).resolve().parent)) import d09_wire as wire import os_update_executor as executor import os_update_owner as os_owner from release_payload import os_component from test_os_update import candidate def load(name, path): spec = importlib.util.spec_from_file_location(name, path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module fixture = load('os_roundtrip_d09_fixture', SCRIPTS.parent / 'tests/regression/d09-lifecycle.py') client = load('os_roundtrip_client', SCRIPTS / 'd09-lifecycle-client.py') class OwnerClientRoundtrip(unittest.TestCase): def setUp(self): self.f = fixture.Lifecycle('test_inspect_challenge_and_current_head') self.f.setUp() self.addCleanup(self.f.tearDown) self.root = self.f.root / 'payloads' / 'os' self.root.mkdir() (self.root / 'transactions').mkdir() self.verifier = self.root / 'verifier' self.verifier.write_bytes(b'\x7fELFdisposable-verifier-identity') self.verifier.chmod(0o755) preservation = wire.canonical(dict(authority_digest='9' * 64, service_policy='reviewed-maintenance-window', preserve_wallets_data_identity_stop_uninstall=True)) self.plan = candidate() self.plan['preservation_sha256'] = wire.digest(preservation) for name in self.plan['boot']['retained_files']: path = self.root / 'rootfs' / name.lstrip('/') path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(('retained:' + name).encode()) self.plan['boot']['retained_files'][name] = wire.digest(path.read_bytes()) raw = wire.canonical(self.plan) self.source = self.root / 'plans' / wire.digest(raw) self.source.mkdir(parents=True) (self.source / 'plan.json').write_bytes(raw) (self.source / 'preservation.json').write_bytes(preservation) component = os_component(self.source / 'plan.json', '2.0.1-alpha', 'https://fixture.invalid') components = [ {**component, 'name': 'archipelago', 'sha256': 'e' * 64, 'size_bytes': 1, 'download_url': 'https://fixture.invalid/v2.0.1-alpha/archipelago'}, {**component, 'name': 'archipelago-frontend-2.0.1-alpha.tar.gz', 'sha256': 'f' * 64, 'size_bytes': 1, 'download_url': 'https://fixture.invalid/v2.0.1-alpha/archipelago-frontend-2.0.1-alpha.tar.gz'}, component, ] self.manifest = dict(version='2.0.1-alpha', release_date='2026-11-03', changelog=[], os_update=dict(version=1, plan_sha256=wire.digest(raw)), components=components) # Fixed disposable PKCS#8 test seed, never a production release key. self.key = self.root / 'fixture-key.der' self.key.write_bytes(bytes.fromhex('302e020100300506032b657004220420') + bytes([7]) * 32) self.public = self.root / 'fixture-public.der' self.openssl('pkey', '-inform', 'DER', '-in', str(self.key), '-pubout', '-outform', 'DER', '-out', str(self.public)) key_bytes = b'\xed\x01' + self.public.read_bytes()[-32:] alphabet = '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz' number, encoded = int.from_bytes(key_bytes, 'big'), '' while number: number, digit = divmod(number, 58) encoded = alphabet[digit] + encoded payload = self.root / 'signing-payload' payload.write_bytes(wire.canonical(self.manifest)) sig = self.root / 'fixture-signature' self.openssl('pkeyutl', '-sign', '-keyform', 'DER', '-inkey', str(self.key), '-rawin', '-in', str(payload), '-out', str(sig)) self.manifest.update(signature=sig.read_bytes().hex(), signed_by='did:key:z' + encoded) manifest_raw = json.dumps(self.manifest, indent=2).encode() (self.source / 'manifest.json').write_bytes(manifest_raw) self.binding = self.f.binding self.binding.update(artifacts=dict(backend='e' * 64, frontend='f' * 64), restore_artifacts=dict(backend='a' * 64, frontend='b' * 64)) self.binding['release_identity'] = ['2.0.1-alpha', sorted([[c[k] for k in ('name', 'new_version', 'sha256', 'size_bytes')] for c in components])] self.binding['os_update'] = dict(version=1, plan_sha256=wire.digest(raw), manifest_sha256=wire.digest(manifest_raw), authority_digest='9' * 64, verifier_sha256=wire.digest(self.verifier.read_bytes())) self.binding['prerequisite']['backend_sha256'] = self.binding['os_update']['verifier_sha256'] incoming = self.root / 'incoming' incoming.mkdir() for name in ('plan.json', 'manifest.json', 'preservation.json'): (incoming / name).write_bytes((self.source / name).read_bytes()) (incoming / 'verifier').write_bytes(self.verifier.read_bytes()) self.transaction = self.root / 'transactions' / ('9' * 64) self.transaction.mkdir() modules = self.root / 'installed' modules.mkdir() for name in os_owner.MODULES: (modules / name).write_bytes((SCRIPTS / name).read_bytes()) for path in [self.root, *self.root.rglob('*')]: path.chmod(0o700 if path.is_dir() or path == self.verifier else 0o600) self.running = 'old' self.kernel_installed = False self.boot = 'new-boot' self.verifications = 0 for module in (executor, os_owner): for name, value in (('ROOT', self.root), ('VERIFIER', self.verifier), ('secure', fixture.fixture_secure), ('command', self.command)): p = patch.object(module, name, value) p.start() self.addCleanup(p.stop) p = patch.object(os_owner.platform, 'release', side_effect=lambda: self.running) p.start() self.addCleanup(p.stop) for name, replacement in ( ('boot_id', lambda: self.boot), ('file_hash', lambda path: wire.digest((self.root / 'rootfs' / str(path).lstrip('/')).read_bytes())), ): p = patch.object(executor, name, replacement) p.start() self.addCleanup(p.stop) p = patch.object(os_owner, 'boot_id', lambda: self.boot) p.start() self.addCleanup(p.stop) p = patch.object(executor.platform, 'version', return_value='fixture Debian 1 test build') p.start() self.addCleanup(p.stop) p = patch.object(os_owner, '__file__', str(modules / 'os_update_owner.py')) p.start() self.addCleanup(p.stop) # qualify checks presence only; no package lock is acquired by inspection. p = patch.dict(sys.modules, {'apt_pkg': types.SimpleNamespace(SystemLock=object)}) p.start() self.addCleanup(p.stop) os_owner.install_inputs(self.binding, incoming) os_owner.qualify(self.binding) def openssl(self, *args): return subprocess.run(['openssl', *args], check=True, capture_output=True) def command(self, args): if args[0] == str(self.verifier): self.assertEqual(args[1:3], ['ceremony', 'verify']) document = json.loads(Path(args[3]).read_bytes()) signature = bytes.fromhex(document.pop('signature')) self.assertEqual(document.pop('signed_by'), self.manifest['signed_by']) (self.root / 'verify-payload').write_bytes(wire.canonical(document)) (self.root / 'verify-signature').write_bytes(signature) self.openssl('pkeyutl', '-verify', '-pubin', '-keyform', 'DER', '-inkey', str(self.public), '-rawin', '-in', str(self.root / 'verify-payload'), '-sigfile', str(self.root / 'verify-signature')) self.verifications += 1 return '' if args == ['/usr/bin/dpkg', '--audit']: return '' if args == ['/usr/bin/dpkg-query', '-W', '-f=' + executor.INVENTORY]: return 'base-files\t13.1\tamd64\tii \nlinux-image-new\t1\tamd64\tii \n' self.assertEqual(args[:3], ['/usr/bin/dpkg-query', '-W', '-f=${Version}\t${db:Status-Abbrev}']) if args[-1] == 'base-files': return '13.1\tii ' self.assertEqual(args[-1], self.plan['boot']['package']['name']) if not self.kernel_installed: raise subprocess.CalledProcessError(1, args) return self.plan['boot']['package']['version'] + '\tii ' def journal(self, phase): # Actual executor's synced journal writer, loaded again by each inspect. e = executor.Executor(self.binding['os_update'], create=False) e.journal(phase) self.f.o.store.checkpoint_os(e.status) def inspect_response(self): before = self.f.o.store.raw request = dict(schema_version=1, transaction_id=None, target_binding_digest=self.binding['target_binding_digest'], challenge='d' * 64) result = self.f.o.inspect(wire.decode('inspect_request', wire.canonical(request)), self.f.peer) bound = wire.bind_inspect(request, wire.decode('inspect_response', wire.canonical(result))) self.assertEqual(self.f.o.store.raw, before, 'inspection must not consume or rewrite authority') return bound def inspect(self): return client.projection(self.binding, self.inspect_response())['os_update'] def test_actual_owner_journal_wire_client_retains_partial_failure(self): observed = self.inspect() self.assertTrue(observed['supported']) self.assertEqual(observed['phase'], 'NO_OS_TRANSACTION') self.assertIsNone(observed['installed_kernel']) self.assertEqual(observed['running_kernel'], 'old') for phase in ('PREPARE', 'DOWNLOAD', 'VERIFIED', 'APPLY_INTENT', 'VERIFY'): self.journal(phase) observed = self.inspect() self.assertEqual(observed['phase'], phase) self.assertFalse(observed['post_boot_verified']) self.assertIsNone(observed['reboot_required']) self.kernel_installed = True self.journal('REBOOT_REQUIRED') observed = self.inspect() self.assertEqual(observed['installed_kernel'], 'new') self.assertEqual(observed['running_kernel'], 'old') self.assertTrue(observed['reboot_required']) self.assertFalse(observed['post_boot_verified']) self.journal('RECONCILIATION_REQUIRED') raw = (self.transaction / 'status.json').read_bytes() self.assertEqual(self.inspect()['phase'], 'RECONCILIATION_REQUIRED') self.assertEqual(self.inspect()['phase'], 'RECONCILIATION_REQUIRED') self.assertEqual((self.transaction / 'status.json').read_bytes(), raw) self.assertGreater(self.verifications, 5) def test_app_only_owner_inspection_remains_legacy_without_os_completion(self): self.binding.pop('os_update') self.binding['release_identity'] = ['B', [['archipelago', 'B', 'e' * 64, 1]]] before = self.f.o.store.raw self.assertIsNone(self.inspect()) self.assertEqual(self.f.o.store.raw, before) def test_actual_client_prepare_acceptance_and_uncertain_retry_preserve_grant_and_request(self): cache = self.root / 'client-cache' cache.mkdir(mode=0o700) request = cache / 'request.json' # Non-OS object appointments belong to the existing baseline fixture; # run the real OS qualification as well before accepting this request. self.f.o.preflight = lambda: os_owner.qualify(self.binding) with patch.object(client, 'CACHE', cache), patch.object(client, 'inspect', side_effect=lambda _: self.inspect_response()), patch.object(client, 'exchange', side_effect=lambda req, _: self.f.submit(wire.decode('submit_request', wire.canonical(req)))): accepted = client.submit_cached(self.binding, 'prepare', 'UPDATER', '5' * 32, 'fixture', request, 'a' * 32) self.assertEqual(accepted['status'], 'ACCEPTED') original_request = request.read_bytes() selected = json.loads(original_request)['selected_release_digest'] self.assertEqual(selected, wire.release_digest(self.binding['release_identity'])) grants = list(self.f.o.store.value['grants']) for phase in ('PREPARE', 'DOWNLOAD', 'VERIFIED', 'APPLY_INTENT', 'RECONCILIATION_REQUIRED'): self.journal(phase) retained = self.f.o.store.raw with self.assertRaisesRegex(PermissionError, 'reconciliation'): client.submit_cached(self.binding, 'prepare', 'UPDATER', '5' * 32, 'fixture', request, 'a' * 32) self.assertEqual(request.read_bytes(), original_request) self.assertEqual(self.f.o.store.value['grants'], grants) self.assertEqual(self.f.o.store.raw, retained) def test_signed_manifest_or_selected_identity_drift_clears_success(self): self.kernel_installed = True for phase in ('PREPARE', 'DOWNLOAD', 'VERIFIED', 'APPLY_INTENT', 'VERIFY', 'REBOOT_REQUIRED'): self.journal(phase) self.assertTrue(self.inspect()['reboot_required']) # Altered bytes refuse even if an attacker rewrites only the outer hash; # the actual signature verifier still covers the plan/release component. original = wire.canonical(self.manifest) for index in range(3): with self.subTest(component=index): self.manifest = json.loads(original) self.manifest['components'][index]['sha256'] = '0' * 64 raw = wire.canonical(self.manifest) (self.source / 'manifest.json').write_bytes(raw) self.binding['os_update']['manifest_sha256'] = wire.digest(raw) observed = self.inspect() self.assertEqual(observed['phase'], 'UNKNOWN') self.assertFalse(observed['post_boot_verified']) self.assertIsNone(observed['reboot_required']) def test_actual_postboot_confirmation_is_rechecked_on_every_inspection(self): self.kernel_installed = True image = self.root / 'rootfs/boot/vmlinuz-new' image.write_bytes(b'qualified fixture kernel') before = self.transaction / 'before-inventory' before.write_text(self.command(['/usr/bin/dpkg-query', '-W', '-f=' + executor.INVENTORY])) before.chmod(0o600) observation = dict(version=1, authority_digest=self.binding['os_update']['authority_digest'], plan_sha256=self.binding['os_update']['plan_sha256'], boot_id=self.boot, preservation_verified=True) path = self.source / 'postboot-preservation.json' path.write_bytes(wire.canonical(observation)) path.chmod(0o600) for phase in ('PREPARE', 'DOWNLOAD', 'VERIFIED', 'APPLY_INTENT', 'VERIFY'): self.journal(phase) e = executor.Executor(self.binding['os_update'], create=False) e.journal('REBOOT_REQUIRED', boot_id='old-boot', boot=dict(release='new', package_version='1', files={'/boot/vmlinuz-new': wire.digest(image.read_bytes())})) self.f.o.store.checkpoint_os(e.status) self.running = 'new' reply = self.f.o.worker_call({'method': 'os_postboot'}, (123, 0, 0)) self.assertTrue(reply['os_postboot_verified']) raw = (self.transaction / 'status.json').read_bytes() observed = self.inspect() self.assertEqual(observed['phase'], 'POSTBOOT_CONFIRMED') self.assertTrue(observed['post_boot_verified']) self.assertFalse(observed['reboot_required']) self.assertEqual(observed['installed_kernel'], observed['running_kernel']) # Later boot on the old kernel cannot reuse the previous success record. self.running = 'old' self.assertEqual(self.inspect()['phase'], 'REBOOT_REQUIRED') self.assertFalse(self.inspect()['post_boot_verified']) self.assertEqual((self.transaction / 'status.json').read_bytes(), raw) self.running = 'new' image.write_bytes(b'changed boot payload') self.assertFalse(self.inspect()['post_boot_verified']) self.assertEqual((self.transaction / 'status.json').read_bytes(), raw) def test_admitted_release_identity_cannot_substitute_another_plan(self): self.assertEqual(self.inspect()['phase'], 'NO_OS_TRANSACTION') next(c for c in self.binding['release_identity'][1] if c[0] == 'os-update-plan.json')[2] = '0' * 64 observed = self.inspect() self.assertEqual(observed['phase'], 'UNKNOWN') self.assertFalse(observed['post_boot_verified']) self.assertFalse((self.transaction / 'status.json').exists()) def test_reformatting_signed_owner_document_requires_matching_raw_byte_admission(self): self.assertEqual(self.inspect()['phase'], 'NO_OS_TRANSACTION') original = (self.source / 'manifest.json').read_bytes() compact = wire.canonical(self.manifest) self.assertNotEqual(compact, original) (self.source / 'manifest.json').write_bytes(compact) observed = self.inspect() self.assertEqual(observed['phase'], 'UNKNOWN') self.assertFalse(observed['post_boot_verified']) self.assertEqual((self.source / 'manifest.json').read_bytes(), compact) def test_corrupt_retained_journal_is_unknown_without_legacy_upgrade_or_rewrite(self): for phase in ('PREPARE', 'DOWNLOAD', 'VERIFIED', 'APPLY_INTENT'): self.journal(phase) path = self.transaction / 'status.json' path.write_bytes(b'{interrupted retained journal') observed = self.inspect() self.assertEqual(observed['phase'], 'UNKNOWN') self.assertFalse(observed['post_boot_verified']) self.assertIsNone(observed['reboot_required']) self.assertEqual(path.read_bytes(), b'{interrupted retained journal') if __name__ == '__main__': unittest.main()