75 lines
3.6 KiB
Python
75 lines
3.6 KiB
Python
"""Authenticated status grammar; no host/package effects."""
|
|
import importlib.util
|
|
import pathlib
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
|
|
import d09_wire as wire
|
|
|
|
|
|
def response():
|
|
return dict(schema_version=1, status='NO_D09_TRANSACTION', reason='NONE',
|
|
challenge='a' * 64, target_binding_digest='b' * 64,
|
|
admission_epoch_id=None, head_digest=None, head_sequence=None,
|
|
receipt=None, receipt_digest=None, transaction_id=None)
|
|
|
|
|
|
def os_status():
|
|
return dict(version=1, supported=True, plan_sha256='c' * 64,
|
|
phase='REBOOT_REQUIRED', installed_os='Debian trixie',
|
|
installed_kernel='6.12-new', running_kernel='6.12-old',
|
|
reboot_required=True, post_boot_verified=False)
|
|
|
|
|
|
class OsWireTests(unittest.TestCase):
|
|
def test_legacy_and_bound_os_inspection(self):
|
|
r = response()
|
|
request = dict(schema_version=1, target_binding_digest=r['target_binding_digest'],
|
|
challenge=r['challenge'], transaction_id=None)
|
|
wire.bind_inspect(request, r)
|
|
r['os_update'] = os_status()
|
|
wire.bind_inspect(request, wire.decode('inspect_response', wire.canonical(r)))
|
|
r['challenge'] = 'd' * 64
|
|
with self.assertRaises(ValueError):
|
|
wire.bind_inspect(request, r)
|
|
|
|
def test_exact_os_types_fields_and_postboot_evidence(self):
|
|
mutations = [dict(version=2), dict(supported=1), dict(extra=True),
|
|
dict(installed_os='x' * 129), dict(running_kernel='bad\ntext'),
|
|
dict(plan_sha256=None), dict(reboot_required=False),
|
|
dict(post_boot_verified=True), dict(phase='COMPLETE'),
|
|
dict(phase='POSTBOOT_CONFIRMED', post_boot_verified=True, reboot_required=False)]
|
|
for mutation in mutations:
|
|
with self.subTest(mutation=mutation), self.assertRaises(ValueError):
|
|
r = response()
|
|
r['os_update'] = {**os_status(), **mutation}
|
|
wire.decode('inspect_response', wire.canonical(r))
|
|
r['os_update'] = {**os_status(), 'phase': 'POSTBOOT_CONFIRMED',
|
|
'post_boot_verified': True, 'reboot_required': False,
|
|
'running_kernel': '6.12-new'}
|
|
wire.decode('inspect_response', wire.canonical(r))
|
|
|
|
def test_python_client_refuses_before_submit_on_legacy_or_partial_dpkg(self):
|
|
path = pathlib.Path(wire.__file__).with_name('d09-lifecycle-client.py')
|
|
spec = importlib.util.spec_from_file_location('d09_os_client_fixture', path)
|
|
client = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(client)
|
|
binding = dict(release_identity=['2.0.1-alpha', [['os-update-plan.json', '2.0.1-alpha', 'c' * 64, 42]]])
|
|
for status in (None, {**os_status(), 'phase': 'RECONCILIATION_REQUIRED'},
|
|
{**os_status(), 'phase': 'UNKNOWN', 'supported': False, 'plan_sha256': None},
|
|
{**os_status(), 'plan_sha256': 'd' * 64}):
|
|
r = response()
|
|
if status is not None:
|
|
r['os_update'] = status
|
|
with patch.object(client, 'inspect', return_value=r), patch.object(client, 'exchange') as exchange:
|
|
expected = 'reconciliation' if status and status['phase'] == 'UNKNOWN' else ''
|
|
with self.assertRaisesRegex(PermissionError, expected):
|
|
client.submit_cached(binding, 'prepare', 'UPDATER', None, 'fixture', pathlib.Path('/nonexistent'))
|
|
exchange.assert_not_called()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|