185 lines
8.7 KiB
Python
185 lines
8.7 KiB
Python
"""An installed-guest update must be bounded, inspectable and reversible."""
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import zipfile
|
|
|
|
|
|
class GuestUpdateTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.script = Path(__file__).resolve().parents[1] / 'dev/guest-update.py'
|
|
|
|
def setUp(self):
|
|
self.assertTrue(self.script.is_file(), 'guest update helper must exist')
|
|
spec = importlib.util.spec_from_file_location('guest_update', self.script)
|
|
self.mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(self.mod)
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.base = Path(self.tmp.name)
|
|
self.source = self.base / 'source'
|
|
for name, value in {'hydra/__init__.py': '', 'hydra/cli.py': 'NEW = True\n',
|
|
'bin/hydra': '#!/usr/bin/python3\n',
|
|
'image/rootfs/etc/motd': 'Hydra OS\n',
|
|
'image/rootfs/usr/share/hydra/fastfetch.txt': 'hydra logo\n'}.items():
|
|
dest = self.source / name
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_text(value)
|
|
self.root = self.base / 'guest'
|
|
marker = self.root / 'usr/share/hydra/build-manifest.json'
|
|
marker.parent.mkdir(parents=True)
|
|
marker.write_text('{"original": true}\n')
|
|
self.old = self.root / 'usr/lib/hydra/hydra/cli.py'
|
|
self.old.parent.mkdir(parents=True)
|
|
self.old.write_text('OLD = True\n')
|
|
self.bundle = self.base / 'update.pyz'
|
|
|
|
def build(self):
|
|
return self.mod.build_bundle(self.source, self.bundle)
|
|
|
|
def test_dry_run_preserves_installed_files_and_user_disk(self):
|
|
self.build()
|
|
disk = self.root / 'home/user/.local/state/hydra/work/disk.qcow2'
|
|
disk.parent.mkdir(parents=True)
|
|
disk.write_bytes(b'user disk')
|
|
result = self.mod.apply_bundle(self.bundle, self.root)
|
|
self.assertFalse(result['applied'])
|
|
self.assertEqual(self.old.read_text(), 'OLD = True\n')
|
|
self.assertEqual(disk.read_bytes(), b'user disk')
|
|
self.assertFalse((self.root / 'var/lib/hydra/updates').exists())
|
|
|
|
def test_apply_saves_backup_and_preserves_original_image_evidence(self):
|
|
self.build()
|
|
result = self.mod.apply_bundle(self.bundle, self.root, execute=True)
|
|
self.assertTrue(result['applied'])
|
|
self.assertEqual(self.old.read_text(), 'NEW = True\n')
|
|
backup = Path(result['backup'])
|
|
self.assertEqual((backup / 'usr/lib/hydra/hydra/cli.py').read_text(), 'OLD = True\n')
|
|
self.assertEqual(json.loads((self.root / 'usr/share/hydra/build-manifest.json').read_text()), {'original': True})
|
|
self.assertTrue((self.root / 'usr/share/hydra/runtime-update.json').is_file())
|
|
self.assertEqual((self.root / 'usr/share/hydra/fastfetch.txt').read_text(), 'hydra logo\n')
|
|
|
|
def test_branding_preserves_vendor_release_and_backs_up_etc_symlink(self):
|
|
vendor = self.root / 'usr/lib/os-release'
|
|
vendor.parent.mkdir(parents=True, exist_ok=True)
|
|
vendor.write_text('NAME="Gentoo"\nID=gentoo\nVERSION_ID=2.18\n')
|
|
release = self.root / 'etc/os-release'
|
|
release.parent.mkdir(parents=True)
|
|
release.symlink_to('../usr/lib/os-release')
|
|
self.build()
|
|
result = self.mod.apply_bundle(self.bundle, self.root, execute=True)
|
|
self.assertFalse(release.is_symlink())
|
|
self.assertIn('PRETTY_NAME="Hydra OS"', release.read_text())
|
|
self.assertIn('ID_LIKE=gentoo', release.read_text())
|
|
self.assertIn('VERSION_ID=2.18', release.read_text())
|
|
self.assertEqual(vendor.read_text(), 'NAME="Gentoo"\nID=gentoo\nVERSION_ID=2.18\n')
|
|
saved = Path(result['backup']) / 'etc/os-release'
|
|
self.assertTrue(saved.is_symlink())
|
|
self.assertEqual(str(saved.readlink()), '../usr/lib/os-release')
|
|
|
|
def test_non_hydra_root_is_refused_before_changes(self):
|
|
self.build()
|
|
(self.root / 'usr/share/hydra/build-manifest.json').unlink()
|
|
with self.assertRaisesRegex(self.mod.UpdateError, 'Hydra'):
|
|
self.mod.apply_bundle(self.bundle, self.root, execute=True)
|
|
self.assertEqual(self.old.read_text(), 'OLD = True\n')
|
|
|
|
def test_modified_payload_is_refused_before_changes(self):
|
|
self.build()
|
|
other = self.base / 'tampered.pyz'
|
|
with zipfile.ZipFile(self.bundle) as src, zipfile.ZipFile(other, 'w') as dst:
|
|
for name in src.namelist():
|
|
data = b'CORRUPTED\n' if name == 'payload/usr/lib/hydra/hydra/cli.py' else src.read(name)
|
|
dst.writestr(name, data)
|
|
with self.assertRaisesRegex(self.mod.UpdateError, 'checksum'):
|
|
self.mod.apply_bundle(other, self.root, execute=True)
|
|
self.assertEqual(self.old.read_text(), 'OLD = True\n')
|
|
|
|
def test_symlink_destination_cannot_redirect_privileged_update(self):
|
|
self.build()
|
|
outside = self.base / 'outside'
|
|
outside.write_text('KEEP\n')
|
|
self.old.unlink()
|
|
self.old.symlink_to(outside)
|
|
with self.assertRaisesRegex(self.mod.UpdateError, '[Ss]ymlink'):
|
|
self.mod.apply_bundle(self.bundle, self.root, execute=True)
|
|
self.assertEqual(outside.read_text(), 'KEEP\n')
|
|
|
|
def test_bundle_never_overwrites_an_existing_output(self):
|
|
self.bundle.write_bytes(b'keep this artifact')
|
|
with self.assertRaises((FileExistsError, self.mod.UpdateError)):
|
|
self.build()
|
|
self.assertEqual(self.bundle.read_bytes(), b'keep this artifact')
|
|
|
|
def test_self_contained_bundle_applies_without_source_checkout(self):
|
|
self.build()
|
|
result = subprocess.run([sys.executable, str(self.bundle), '--root', str(self.root), '--apply'],
|
|
capture_output=True, text=True, timeout=10, cwd=self.base)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(self.old.read_text(), 'NEW = True\n')
|
|
self.assertIn('Runtime updated', result.stdout)
|
|
|
|
def test_write_failure_rolls_back_files_already_replaced(self):
|
|
self.build()
|
|
actual_write = self.mod.atomic_write
|
|
calls = 0
|
|
|
|
def fail_once(*args):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 3:
|
|
raise OSError('injected full disk')
|
|
return actual_write(*args)
|
|
|
|
with patch.object(self.mod, 'atomic_write', side_effect=fail_once):
|
|
with self.assertRaisesRegex(OSError, 'full disk'):
|
|
self.mod.apply_bundle(self.bundle, self.root, execute=True)
|
|
self.assertEqual(self.old.read_text(), 'OLD = True\n')
|
|
self.assertFalse((self.root / 'etc/motd').exists())
|
|
self.assertFalse((self.root / 'usr/lib/hydra/bin/hydra').exists())
|
|
|
|
def test_persistent_restore_failures_attempt_all_cleanup_and_preserve_apply_error(self):
|
|
import contextlib
|
|
import io
|
|
launcher = self.root / 'usr/lib/hydra/bin/hydra'
|
|
launcher.parent.mkdir(parents=True)
|
|
launcher.write_text('OLD LAUNCHER\n')
|
|
self.build()
|
|
actual_write = self.mod.atomic_write
|
|
original_error = OSError('original apply failure: disk remains full')
|
|
calls = 0
|
|
|
|
def persistent_failure(*args):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 5:
|
|
raise original_error
|
|
if calls > 5:
|
|
raise OSError('persistent restore failure: disk remains full')
|
|
return actual_write(*args)
|
|
|
|
errors = io.StringIO()
|
|
with contextlib.redirect_stderr(errors), patch.object(self.mod, 'atomic_write', side_effect=persistent_failure):
|
|
with self.assertRaises(OSError) as raised:
|
|
self.mod.apply_bundle(self.bundle, self.root, execute=True)
|
|
self.assertIs(raised.exception, original_error)
|
|
self.assertEqual(self.old.read_text(), 'NEW = True\n')
|
|
self.assertFalse((self.root / 'etc/motd').exists())
|
|
self.assertFalse((self.root / 'usr/lib/hydra/hydra/__init__.py').exists())
|
|
backups = list((self.root / 'var/lib/hydra/updates').iterdir())
|
|
self.assertEqual(len(backups), 1)
|
|
backup = backups[0]
|
|
self.assertEqual((backup / 'usr/lib/hydra/hydra/cli.py').read_text(), 'OLD = True\n')
|
|
self.assertEqual((backup / 'usr/lib/hydra/bin/hydra').read_text(), 'OLD LAUNCHER\n')
|
|
self.assertIn(str(backup), errors.getvalue())
|
|
self.assertIn('rollback incomplete', errors.getvalue().lower())
|
|
self.assertIn('usr/lib/hydra/hydra/cli.py', errors.getvalue())
|
|
self.assertIn('usr/lib/hydra/bin/hydra', errors.getvalue())
|
|
self.assertIn('persistent restore failure', errors.getvalue())
|