Add read-only laptop doctor and trusted image recovery entrypoints
This commit is contained in:
Codex
@@ -21,6 +21,29 @@ class CLITests(unittest.TestCase):
|
||||
def run_cli(self, *args):
|
||||
return subprocess.run(self.base + list(args), capture_output=True, text=True, timeout=5)
|
||||
|
||||
def test_doctor_json_and_text_do_not_create_runtime_state(self):
|
||||
from unittest.mock import patch
|
||||
from hydra import cli
|
||||
import contextlib
|
||||
import io
|
||||
with patch('hydra.doctor.doctor_report', return_value={'format': 1, 'scope': 'observations',
|
||||
'manual_checks': ['Wi-Fi connection', 'suspend']}):
|
||||
output = io.StringIO()
|
||||
with contextlib.redirect_stdout(output):
|
||||
try:
|
||||
status = cli.main(self.base[2:] + ['doctor', '--json'])
|
||||
except SystemExit as error:
|
||||
status = error.code
|
||||
self.assertEqual(status, 0)
|
||||
self.assertEqual(json.loads(output.getvalue())['format'], 1)
|
||||
output = io.StringIO()
|
||||
with contextlib.redirect_stdout(output):
|
||||
self.assertEqual(cli.main(self.base[2:] + ['doctor']), 0)
|
||||
self.assertIn('Manual checks', output.getvalue())
|
||||
self.assertFalse((self.root / 'state').exists())
|
||||
self.assertFalse((self.root / 'config').exists())
|
||||
self.assertFalse((self.root / 'run').exists())
|
||||
|
||||
def test_defaults_affect_new_profiles_only_and_explicit_options_win(self):
|
||||
self.defaults.write_text('network="bridge"\nbridge="hydrabr0"\nwidth=1920\nheight=1080\nrefresh_millihz=144000\n')
|
||||
added = self.run_cli('add', 'first', '--width', '1600', '--network', 'user', '--refresh-millihz', '0')
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
try:
|
||||
from hydra import doctor
|
||||
except ImportError:
|
||||
doctor = None
|
||||
|
||||
|
||||
class DoctorTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.assertIsNotNone(doctor, 'read-only doctor module is missing')
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.root = Path(self.tmp.name)
|
||||
for name in ('proc', 'sys', 'dev'):
|
||||
(self.root / name).mkdir()
|
||||
|
||||
def file(self, name, value=''):
|
||||
path = self.root / name
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(value)
|
||||
return path
|
||||
|
||||
def report(self, **kwargs):
|
||||
return doctor.doctor_report(proc_root=self.root / 'proc', sys_root=self.root / 'sys',
|
||||
dev_root=self.root / 'dev', find_tool=lambda _: None, **kwargs)
|
||||
|
||||
def test_absent_hardware_is_reported_without_false_functional_claims(self):
|
||||
result = self.report()
|
||||
self.assertEqual(result['firmware']['uefi'], False)
|
||||
self.assertIsNone(result['firmware']['secure_boot'])
|
||||
self.assertEqual(result['storage']['devices'], [])
|
||||
self.assertFalse(result['kvm']['device_present'])
|
||||
self.assertTrue(all(not value for value in result['tools'].values()))
|
||||
self.assertIn('Wi-Fi connection', ' '.join(result['manual_checks']))
|
||||
self.assertIn('144 Hz', ' '.join(result['manual_checks']))
|
||||
json.dumps(result)
|
||||
|
||||
def test_fake_hardware_reports_presence_and_driver_without_identifiers(self):
|
||||
self.file('proc/cpuinfo', 'processor : 0\nflags : vmx sse\nSerial : PRIVATE_SERIAL\n')
|
||||
self.file('proc/meminfo', 'MemTotal: 8388608 kB\nMemAvailable: 4194304 kB\n')
|
||||
secure = self.file('sys/firmware/efi/efivars/SecureBoot-8be4df61-93ca-11d2-aa0d-00e098032b8c')
|
||||
secure.write_bytes(b'\x07\0\0\0\x01')
|
||||
self.file('dev/kvm')
|
||||
self.file('sys/module/kvm_intel/refcnt', '0')
|
||||
self.file('sys/class/block/nvme0n1/size', '2097152')
|
||||
self.file('sys/class/block/nvme0n1/device/serial', 'PRIVATE_SERIAL')
|
||||
self.file('sys/class/block/nvme0n1p1/partition', '1')
|
||||
self.file('sys/class/drm/card0-eDP-1/status', 'connected')
|
||||
self.file('sys/class/drm/card0-eDP-1/modes', '1920x1080\nPRIVATE_USER\n')
|
||||
self.file('sys/bus/pci/devices/0000:00:02.0/class', '0x030000')
|
||||
driver = self.root / 'sys/bus/pci/drivers/i915'
|
||||
driver.mkdir(parents=True)
|
||||
(self.root / 'sys/bus/pci/devices/0000:00:02.0/driver').symlink_to(driver)
|
||||
self.file('sys/class/net/wlan0/wireless/present')
|
||||
self.file('sys/class/net/wlan0/address', 'aa:bb:cc:dd:ee:ff')
|
||||
self.file('sys/class/net/wlan0/ssid', 'PRIVATE_SSID')
|
||||
self.file('sys/class/sound/card0/id', 'PRIVATE_USER')
|
||||
self.file('sys/class/input/event0/device/name', 'PRIVATE_USER')
|
||||
self.file('sys/class/input/event0/device/capabilities/key', hex((1 << 30) | (1 << 44))[2:])
|
||||
result = self.report()
|
||||
self.assertEqual(result['cpu']['virtualization_flags'], ['vmx'])
|
||||
self.assertEqual(result['memory']['total_mib'], 8192)
|
||||
self.assertEqual(result['firmware']['secure_boot'], True)
|
||||
self.assertEqual(result['storage']['devices'][0]['size_bytes'], 1073741824)
|
||||
self.assertEqual(result['graphics']['drivers'], ['i915'])
|
||||
self.assertEqual(result['network']['wireless_devices'], 1)
|
||||
self.assertEqual(result['audio']['cards'], 1)
|
||||
self.assertEqual(result['input']['keyboards'], 1)
|
||||
encoded = json.dumps(result)
|
||||
for secret in ('PRIVATE_SERIAL', 'PRIVATE_USER', 'PRIVATE_SSID', 'aa:bb:cc:dd:ee:ff'):
|
||||
self.assertNotIn(secret, encoded)
|
||||
|
||||
def test_probes_timeout_independently_and_discard_untrusted_output(self):
|
||||
calls = []
|
||||
def run(command, **kwargs):
|
||||
calls.append((command, kwargs))
|
||||
if len(calls) == 1:
|
||||
raise subprocess.TimeoutExpired(command, kwargs['timeout'], output='PRIVATE_SSID')
|
||||
return subprocess.CompletedProcess(command, 5, 'PRIVATE_USER 192.0.2.1 aa:bb:cc:dd:ee:ff', '')
|
||||
result = doctor.doctor_report(proc_root=self.root / 'proc', sys_root=self.root / 'sys',
|
||||
dev_root=self.root / 'dev', find_tool=lambda name: '/usr/bin/' + name,
|
||||
run=run)
|
||||
self.assertIn('timeout', result['services'].values())
|
||||
self.assertIn('unknown', result['services'].values())
|
||||
self.assertNotIn('PRIVATE', json.dumps(result))
|
||||
self.assertLessEqual(len(calls), 8)
|
||||
for argv, options in calls:
|
||||
self.assertEqual(argv[0], '/usr/bin/systemctl')
|
||||
self.assertIn('is-active', argv)
|
||||
self.assertLessEqual(options['timeout'], 1)
|
||||
self.assertFalse(options.get('shell', False))
|
||||
|
||||
def test_reads_are_bounded_and_invalid_numbers_are_unknown(self):
|
||||
self.file('proc/meminfo', 'MemTotal: invalid kB\n' + 'secret' * 100000)
|
||||
self.file('sys/class/block/sda/size', '-1')
|
||||
self.file('sys/class/drm/card0-HDMI-A-1/status', 'private text')
|
||||
result = self.report()
|
||||
self.assertIsNone(result['memory']['total_mib'])
|
||||
self.assertIsNone(result['storage']['devices'][0]['size_bytes'])
|
||||
self.assertEqual(result['graphics']['connectors'][0]['status'], 'unknown')
|
||||
self.assertLess(len(json.dumps(result)), 10000)
|
||||
|
||||
def test_text_and_json_outputs_share_report_and_manual_limits(self):
|
||||
with patch.object(doctor, 'doctor_report', return_value=self.report()):
|
||||
output = io.StringIO()
|
||||
with contextlib.redirect_stdout(output):
|
||||
self.assertEqual(doctor.doctor_main(['--json']), 0)
|
||||
self.assertEqual(json.loads(output.getvalue())['firmware']['uefi'], False)
|
||||
output = io.StringIO()
|
||||
with contextlib.redirect_stdout(output):
|
||||
self.assertEqual(doctor.doctor_main([]), 0)
|
||||
self.assertIn('Manual checks', output.getvalue())
|
||||
self.assertIn('suspend', output.getvalue())
|
||||
@@ -35,6 +35,66 @@ class ImageTests(unittest.TestCase):
|
||||
self.file('boot/vmlinuz-' + version, 'kernel')
|
||||
return version
|
||||
|
||||
def test_installs_independent_updater_and_base_recovery_assets(self):
|
||||
self.kernel()
|
||||
source = Path(__file__).resolve().parents[1]
|
||||
image.configure_rootfs(self.root, source)
|
||||
for relative, asset, mode in (
|
||||
('usr/lib/hydra-update/engine.py', 'hydra/update_engine.py', 0o644),
|
||||
('usr/bin/hydra-update', 'bin/hydra-update', 0o755),
|
||||
('usr/bin/hydra-update-recover', 'bin/hydra-update-recover', 0o755),
|
||||
('usr/lib/hydra/bin/hydra-system', 'bin/hydra-system', 0o755),
|
||||
('etc/hydra/update.pub', 'image/rootfs/etc/hydra/update.pub', 0o644),
|
||||
):
|
||||
with self.subTest(relative=relative):
|
||||
path = self.root / relative
|
||||
self.assertTrue(path.is_file(), relative)
|
||||
self.assertEqual(path.read_bytes(), (source / asset).read_bytes())
|
||||
self.assertEqual(path.stat().st_mode & 0o777, mode)
|
||||
wrapper = self.root / 'usr/bin/hydra-system'
|
||||
self.assertTrue(wrapper.is_file(), 'base recovery launcher is missing')
|
||||
self.assertEqual(wrapper.stat().st_mode & 0o777, 0o755)
|
||||
self.assertEqual(wrapper.read_text(),
|
||||
'#!/usr/bin/python3 -I\nimport sys\nsys.path.insert(0, "/usr/lib/hydra")\n'
|
||||
'from hydra.system_recovery import main\nraise SystemExit(main())\n')
|
||||
self.assertEqual((self.root / 'usr/lib/hydra/hydra/doctor.py').read_bytes(),
|
||||
(source / 'hydra/doctor.py').read_bytes())
|
||||
|
||||
def test_root_assembly_sets_trusted_asset_ownership(self):
|
||||
from unittest.mock import patch
|
||||
self.kernel()
|
||||
with patch('hydra.image.os.geteuid', return_value=0), patch('hydra.image.os.chown') as chown:
|
||||
image.configure_rootfs(self.root, Path(__file__).resolve().parents[1])
|
||||
for relative in ('usr/lib/hydra-update/engine.py', 'usr/lib/hydra-update',
|
||||
'usr/bin/hydra-update', 'usr/bin/hydra-update-recover', 'etc/hydra/update.pub'):
|
||||
chown.assert_any_call(self.root / relative, 0, 0)
|
||||
|
||||
def test_missing_trusted_asset_refuses_configuration_before_mutation(self):
|
||||
import shutil
|
||||
source = Path(__file__).resolve().parents[1]
|
||||
fixture = Path(self.tmp.name) / 'source'
|
||||
fixture.mkdir()
|
||||
for directory in ('bin', 'hydra', 'image'):
|
||||
shutil.copytree(source / directory, fixture / directory,
|
||||
ignore=shutil.ignore_patterns('__pycache__'))
|
||||
for asset in ('bin/hydra-update', 'bin/hydra-update-recover', 'hydra/update_engine.py',
|
||||
'bin/hydra-system', 'hydra/system_recovery.py', 'image/rootfs/etc/hydra/update.pub'):
|
||||
with self.subTest(asset=asset):
|
||||
self.root = Path(self.tmp.name) / ('root-' + asset.replace('/', '-'))
|
||||
self.root.mkdir()
|
||||
self.kernel('6.18.54-gentoo-dist-bin')
|
||||
original = self.file('etc/machine-id', 'preserved-builder-id')
|
||||
missing = fixture / asset
|
||||
saved = missing.read_bytes()
|
||||
missing.unlink()
|
||||
try:
|
||||
with self.assertRaisesRegex(image.ImageError, 'missing'):
|
||||
image.configure_rootfs(self.root, fixture)
|
||||
self.assertEqual(original.read_text(), 'preserved-builder-id')
|
||||
self.assertFalse((self.root / 'usr').exists())
|
||||
finally:
|
||||
missing.write_bytes(saved)
|
||||
|
||||
def test_kernel_from_payload_not_running_host(self):
|
||||
self.kernel()
|
||||
self.assertEqual(image.kernel_version(self.root), '6.12.58-gentoo-dist')
|
||||
|
||||
Reference in New Issue
Block a user