346 lines
18 KiB
Python
346 lines
18 KiB
Python
"""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()
|