729 lines
37 KiB
Python
729 lines
37 KiB
Python
"""Regression tests for disposable-VM isolation and trustworthy serial evidence."""
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import socket
|
|
import select
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / 'dev/iso-smoke.py'
|
|
|
|
|
|
class SmokeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.assertTrue(SCRIPT.is_file(), 'disposable ISO acceptance harness is missing')
|
|
spec = importlib.util.spec_from_file_location('iso_smoke', SCRIPT)
|
|
self.mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(self.mod)
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.code = self.root / 'code.fd'
|
|
self.variables = self.root / 'vars.fd'
|
|
self.code.write_bytes(b'firmware')
|
|
self.variables.write_bytes(b'fresh vars')
|
|
|
|
def test_existing_disk_or_symlink_cannot_be_a_test_drive(self):
|
|
disk = self.root / 'disk.qcow2'
|
|
disk.write_bytes(b'original')
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.reserve_disk(disk)
|
|
link = self.root / 'link.qcow2'
|
|
link.symlink_to(disk)
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.reserve_disk(link)
|
|
self.assertEqual(disk.read_bytes(), b'original')
|
|
|
|
def test_qemu_only_accepts_owned_regular_disk_and_readonly_iso(self):
|
|
disk = self.root / 'disk.qcow2'
|
|
self.mod.reserve_disk(disk)
|
|
iso = self.root / 'trusted.iso'
|
|
iso.write_bytes(b'ISO')
|
|
argv = self.mod.qemu_argv('qemu', self.root, disk, self.code,
|
|
self.variables, iso, 2048)
|
|
drives = [argv[i + 1] for i, value in enumerate(argv[:-1]) if value == '-drive']
|
|
self.assertTrue(any('format=qcow2' in item and str(disk) in item for item in drives))
|
|
self.assertTrue(any('media=cdrom' in item and 'readonly=on' in item for item in drives))
|
|
self.assertIn('accel=kvm', ' '.join(argv))
|
|
disk.unlink()
|
|
disk.symlink_to(iso)
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.qemu_argv('qemu', self.root, disk, self.code,
|
|
self.variables, iso, 2048)
|
|
|
|
def test_installed_reboot_has_no_optical_medium(self):
|
|
disk = self.root / 'disk.qcow2'
|
|
self.mod.reserve_disk(disk)
|
|
argv = self.mod.qemu_argv('qemu', self.root, disk, self.code,
|
|
self.variables, None, 2048)
|
|
self.assertNotIn('media=cdrom', ' '.join(argv))
|
|
self.assertNotIn('-cdrom', argv)
|
|
|
|
def test_persistent_optical_drive_keeps_topology_with_no_installed_medium(self):
|
|
disk = self.root / 'disk.qcow2'
|
|
self.mod.reserve_disk(disk)
|
|
iso = self.root / 'trusted.iso'
|
|
iso.write_bytes(b'ISO')
|
|
optical = []
|
|
for medium in (iso, None):
|
|
argv = self.mod.qemu_argv('qemu', self.root, disk, self.code,
|
|
self.variables, medium, 2048, keep_cdrom_drive=True)
|
|
drives = [argv[i + 1] for i, value in enumerate(argv[:-1]) if value == '-drive']
|
|
cdrom = [dict(field.split('=', 1) for field in drive.split(','))
|
|
for drive in drives if 'media=cdrom' in drive]
|
|
self.assertEqual(len(cdrom), 1)
|
|
self.assertEqual(cdrom[0]['if'], 'ide')
|
|
self.assertEqual(cdrom[0]['index'], '0')
|
|
self.assertEqual(cdrom[0]['id'], 'acceptance-cdrom')
|
|
self.assertEqual(cdrom[0]['readonly'], 'on')
|
|
optical.append(cdrom[0])
|
|
self.assertEqual(optical[0].pop('file'), str(iso))
|
|
self.assertEqual(optical[0].pop('format'), 'raw')
|
|
self.assertNotIn('file', optical[1], 'Installed boot must have no ISO medium')
|
|
self.assertNotIn('format', optical[1], 'An empty drive must not request a raw block backend')
|
|
self.assertEqual(optical[0], optical[1])
|
|
|
|
def test_hardlinked_disk_is_refused(self):
|
|
disk = self.root / 'disk.qcow2'
|
|
self.mod.reserve_disk(disk)
|
|
(self.root / 'alias').hardlink_to(disk)
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.qemu_argv('qemu', self.root, disk, self.code,
|
|
self.variables, None, 2048)
|
|
|
|
def test_result_frame_ignores_command_echo_and_preserves_failure(self):
|
|
command = self.mod.frame_command('printf sentinel; false', 'testtoken')
|
|
result = subprocess.run(['/bin/sh', '-c', command], capture_output=True)
|
|
echo = command.encode() + b'\r\n'
|
|
self.assertIsNone(self.mod.parse_frame(echo, 'testtoken'))
|
|
self.assertEqual(self.mod.parse_frame(echo + result.stdout, 'testtoken'),
|
|
(1, b'sentinel'))
|
|
|
|
def test_console_redacts_password_even_when_split_across_reads(self):
|
|
one, two = socket.socketpair()
|
|
self.addCleanup(one.close)
|
|
self.addCleanup(two.close)
|
|
log = self.root / 'serial.log'
|
|
console = self.mod.Console(one, log, ['split-secret'])
|
|
self.addCleanup(console.close)
|
|
two.sendall(b'hello split-')
|
|
console.read(0.1)
|
|
two.sendall(b'secret goodbye')
|
|
console.read(0.1)
|
|
console.close()
|
|
self.assertNotIn(b'split-secret', log.read_bytes())
|
|
self.assertIn(b'[REDACTED]', log.read_bytes())
|
|
|
|
def test_console_timeout_is_bounded_and_partial_frame_not_success(self):
|
|
one, two = socket.socketpair()
|
|
self.addCleanup(one.close)
|
|
self.addCleanup(two.close)
|
|
console = self.mod.Console(one, self.root / 'serial.log', [])
|
|
self.addCleanup(console.close)
|
|
two.sendall(b'\\036token:END:0\\037')
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
console.command('true', timeout=0.05)
|
|
|
|
def installer_console(self):
|
|
one, two = socket.socketpair()
|
|
self.addCleanup(two.close)
|
|
console = self.mod.Console(one, self.root / 'installer.log', ['test-secret'])
|
|
self.addCleanup(console.close)
|
|
return console, two
|
|
|
|
def test_captured_installer_refusal_fails_before_password_timeout(self):
|
|
console, guest = self.installer_console()
|
|
guest.sendall(b'\x1b]3008;type=command\x1b\\hydra-install: Command failed (1): '
|
|
b'findmnt --json --target /LiveOS/squashfs.img '
|
|
b'--output SOURCE,TARGET,FSTYPE,MAJ:MIN,OPTIONS: \r\n')
|
|
with self.assertRaisesRegex(self.mod.SmokeError, 'Installer.*findmnt'):
|
|
console.expect_installer(rb'Installed account password: ', 0.05, 'testtoken')
|
|
|
|
def test_installer_shell_return_without_completion_is_failure_even_zero(self):
|
|
for status in (0, 1, 127):
|
|
with self.subTest(status=status):
|
|
console, guest = self.installer_console()
|
|
guest.sendall(b'\x1etesttoken:INSTALL_EXIT:' + str(status).encode() + b'\x1f')
|
|
with self.assertRaisesRegex(self.mod.SmokeError, 'Installer.*INSTALL_EXIT'):
|
|
console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
|
|
0.05, 'testtoken')
|
|
|
|
def test_installer_echo_cannot_fake_exit_and_complete_prompt_still_required(self):
|
|
console, guest = self.installer_console()
|
|
guest.sendall(self.mod.installer_command('testtoken').encode() + b'\r\n'
|
|
b'Installed account password: ')
|
|
self.assertEqual(console.expect_installer(rb'Installed account password: ',
|
|
0.05, 'testtoken').group(), b'Installed account password: ')
|
|
guest.sendall(b'Installation completed. Diagnostic')
|
|
with self.assertRaisesRegex(self.mod.SmokeError, 'Timed out'):
|
|
console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
|
|
0.05, 'testtoken')
|
|
guest.sendall(b' log: /var/log/hydra-install.log\r\n')
|
|
self.assertIn(b'/var/log/hydra-install.log', console.expect_installer(
|
|
rb'Installation completed\. Diagnostic log: [^\r\n]+', 0.05, 'testtoken').group())
|
|
|
|
def test_installer_wrapper_marks_actual_shell_return_and_redacts_error(self):
|
|
console, guest = self.installer_console()
|
|
fake_sudo = self.root / 'sudo'
|
|
fake_sudo.write_text('#!/bin/sh\nprintf "hydra-install: test-secret refused\\n"\nexit 1\n')
|
|
fake_sudo.chmod(0o700)
|
|
result = subprocess.run(['/bin/sh', '-c', self.mod.installer_command('testtoken')],
|
|
env={'PATH': str(self.root)}, capture_output=True, check=True)
|
|
self.assertIn(b'\x1etesttoken:INSTALL_EXIT:1\x1f', result.stdout)
|
|
guest.sendall(result.stdout)
|
|
with self.assertRaisesRegex(self.mod.SmokeError, r'Installer.*\[REDACTED\] refused') as caught:
|
|
console.expect_installer(rb'Repeat password: ', 0.05, 'testtoken')
|
|
self.assertNotIn('test-secret', str(caught.exception))
|
|
|
|
def test_installer_followup_waits_for_exit_and_new_shell_prompt(self):
|
|
console, guest = self.installer_console()
|
|
premature = []
|
|
def respond():
|
|
guest.sendall(b'Installation completed. Diagnostic log: /run/installer.log\r\n')
|
|
premature.append(bool(select.select([guest], [], [], 0.05)[0]))
|
|
guest.sendall(b'\x1etesttoken:INSTALL_EXIT:0\x1f')
|
|
premature.append(bool(select.select([guest], [], [], 0.05)[0]))
|
|
guest.sendall(b'\x1b[?2004hhydra-live@hydra-os ~ $ ')
|
|
data = b''
|
|
while not data.endswith(b'\n'):
|
|
data += guest.recv(65536)
|
|
result = subprocess.run(['/bin/sh', '-c', data.decode()], capture_output=True)
|
|
guest.sendall(result.stdout)
|
|
worker = threading.Thread(target=respond, daemon=True)
|
|
worker.start()
|
|
console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+',
|
|
1, 'testtoken')
|
|
console.finish_installer('testtoken', timeout=1)
|
|
self.assertEqual(console.command('printf ready', timeout=1), b'ready')
|
|
worker.join(timeout=1)
|
|
self.assertEqual(premature, [False, False], 'command reached installer before shell was ready')
|
|
|
|
def test_installer_completion_does_not_override_nonzero_return(self):
|
|
console, guest = self.installer_console()
|
|
guest.sendall(b'\x1etesttoken:INSTALL_EXIT:1\x1fhydra-live@hydra-os ~ $ ')
|
|
with self.assertRaisesRegex(self.mod.SmokeError, 'Installer.*status 1'):
|
|
console.finish_installer('testtoken', timeout=0.05)
|
|
|
|
def shutdown_vm(self, program):
|
|
one, two = socket.socketpair()
|
|
vm = self.mod.VM.__new__(self.mod.VM)
|
|
vm.console = self.mod.Console(one, self.root / 'shutdown.log', ['split-secret'])
|
|
self.addCleanup(vm.console.close)
|
|
code = ('import socket,sys,time; s=socket.socket(fileno=int(sys.argv[1])); ' + program)
|
|
vm.process = subprocess.Popen([sys.executable, '-c', code, str(two.fileno())],
|
|
pass_fds=(two.fileno(),), stdin=subprocess.DEVNULL)
|
|
two.close()
|
|
def reap():
|
|
if vm.process.poll() is None:
|
|
vm.process.terminate()
|
|
vm.process.wait(timeout=2)
|
|
self.addCleanup(reap)
|
|
return vm
|
|
|
|
def test_poweroff_drains_and_redacts_serial_until_owned_process_exits(self):
|
|
vm = self.shutdown_vm("s.sendall(b'Stopping split-'); time.sleep(.02); "
|
|
"s.sendall(b'secret service\\nPower down\\n'); s.close()")
|
|
vm.wait_poweroff(1)
|
|
output = (self.root / 'shutdown.log').read_bytes()
|
|
self.assertIn(b'Stopping [REDACTED] service\nPower down\n', output)
|
|
self.assertNotIn(b'split-secret', output)
|
|
|
|
def test_poweroff_preserves_failure_cause_and_original_timeout(self):
|
|
for program, error in (("s.sendall(b'Stop job stuck\\n'); s.close(); time.sleep(1)",
|
|
'Guest failed to power off in time'),
|
|
("s.sendall(b'QEMU shutdown failed\\n'); s.close(); sys.exit(7)",
|
|
'Test QEMU failed during poweroff')):
|
|
with self.subTest(program=program):
|
|
vm = self.shutdown_vm(program)
|
|
started = time.monotonic()
|
|
with self.assertRaisesRegex(self.mod.SmokeError, error):
|
|
vm.wait_poweroff(.15)
|
|
self.assertLess(time.monotonic() - started, .5)
|
|
output = (self.root / 'shutdown.log').read_bytes()
|
|
self.assertIn(b'Stop job stuck' if 'time.sleep' in program else b'QEMU shutdown failed', output)
|
|
if 'time.sleep' in program:
|
|
self.assertIsNone(vm.process.poll(), 'timeout must not kill or falsely pass the owned process')
|
|
|
|
def test_frame_real_shell_roundtrip_does_not_accept_sentinel_in_echo(self):
|
|
one, two = socket.socketpair()
|
|
self.addCleanup(one.close)
|
|
self.addCleanup(two.close)
|
|
console = self.mod.Console(one, self.root / 'serial.log', [])
|
|
self.addCleanup(console.close)
|
|
def guest():
|
|
data = b''
|
|
while not data.endswith(b'\n'):
|
|
data += two.recv(65536)
|
|
two.sendall(data)
|
|
result = subprocess.run(['/bin/sh', '-c', data.decode()], capture_output=True)
|
|
two.sendall(result.stdout)
|
|
thread = threading.Thread(target=guest)
|
|
thread.start()
|
|
self.assertEqual(console.command("printf 'persisted-data'", timeout=2), b'persisted-data')
|
|
thread.join(timeout=2)
|
|
|
|
def test_firmware_vars_must_be_private_copy_in_scratch(self):
|
|
disk = self.root / 'disk.qcow2'
|
|
self.mod.reserve_disk(disk)
|
|
self.variables.unlink()
|
|
self.variables.symlink_to(self.code)
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.qemu_argv('qemu', self.root, disk, self.code,
|
|
self.variables, None, 2048)
|
|
|
|
def test_drive_outside_scratch_and_ambiguous_qemu_paths_are_refused(self):
|
|
disk = self.root / 'disk.qcow2'
|
|
self.mod.reserve_disk(disk)
|
|
other = self.root / 'other'
|
|
other.mkdir()
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.qemu_argv('qemu', other, disk, self.code,
|
|
self.variables, None, 2048)
|
|
comma = self.root / 'disk,cache=unsafe.qcow2'
|
|
self.mod.reserve_disk(comma)
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.qemu_argv('qemu', self.root, comma, self.code,
|
|
self.variables, None, 2048)
|
|
|
|
def probe_fixture(self, case, live=False):
|
|
# Run the production shell against ordinary scratch files. Commands that
|
|
# query guest state are boundary fixtures; grep/find/test are real tools.
|
|
import os
|
|
root = self.root / 'guest'
|
|
for directory in ('etc/hydra', 'etc/greetd', 'etc/sudoers.d',
|
|
'etc/systemd/system/getty@tty2.service.d',
|
|
'etc/systemd/system/serial-getty@ttyS0.service.d',
|
|
'usr/share/firmware', 'proc', 'sys/firmware/efi',
|
|
'run/initramfs/live/LiveOS'):
|
|
(root / directory).mkdir(parents=True, exist_ok=True)
|
|
(root / 'etc/greetd/config.toml').write_text('normal authentication')
|
|
(root / 'usr/share/firmware/OVMF_CODE.fd').write_text('firmware')
|
|
(root / 'proc/cmdline').write_text('root=UUID=disk ro' if case != 'live-cmdline' else 'root=live:CDLABEL=HYDRA_OS rd.live.image')
|
|
if live or case == 'live-marker':
|
|
(root / 'etc/hydra/live-mode').mkdir(exist_ok=True)
|
|
(root / 'etc/hydra/live-mode/active').touch()
|
|
if live:
|
|
(root / 'run/initramfs/live/LiveOS/squashfs.img').touch()
|
|
files = {
|
|
'live-dropin': ('etc/systemd/system/getty@tty2.service.d/90-hydra-live.conf', 'live'),
|
|
'live-sudoers': ('etc/sudoers.d/90-hydra-live', 'live'),
|
|
'autologin': ('etc/systemd/system/serial-getty@ttyS0.service.d/other.conf', 'ExecStart=agetty --autologin hydratest'),
|
|
'nopasswd': ('etc/sudoers.d/other', 'hydratest ALL=(ALL) NOPASSWD: ALL'),
|
|
'live-greetd': ('etc/greetd/live.toml', 'live'),
|
|
}
|
|
if case in files:
|
|
path, text = files[case]
|
|
(root / path).write_text(text)
|
|
if case == 'config-read-error':
|
|
unreadable = root / 'etc/sudoers.d/unreadable'
|
|
unreadable.write_text('configuration')
|
|
unreadable.chmod(0)
|
|
if case == 'cmdline-read-error':
|
|
(root / 'proc/cmdline').unlink()
|
|
commands = self.root / 'fixture-bin'
|
|
commands.mkdir(exist_ok=True)
|
|
dispatcher = commands / 'fixture'
|
|
dispatcher.write_text('''#!/usr/bin/env python3
|
|
import os, pathlib, sys
|
|
name = pathlib.Path(sys.argv[0]).name
|
|
args = sys.argv[1:]
|
|
case = os.environ['SMOKE_CASE']
|
|
if name == 'id': print('hydra-live')
|
|
elif name == 'findmnt': print('/dev/vda2' if 'SOURCE' in args else ('overlay' if os.environ['SMOKE_LIVE']=='1' else 'ext4'))
|
|
elif name == 'systemctl':
|
|
if args[0] == 'is-active':
|
|
states = [unit != case for unit in args[1:]]
|
|
print('\\n'.join('active' if state else 'failed' for state in states))
|
|
sys.exit(0 if any(states) else 3)
|
|
print('success' if 'Result' in args else '0')
|
|
elif name == 'getent':
|
|
if args[:2] == ['passwd','hydra-live']:
|
|
if case == 'account-lookup-error': sys.exit(3)
|
|
if case != 'live-account': sys.exit(2)
|
|
print('hydra-live:x:1000:1000::/home/hydra-live:/bin/bash')
|
|
elif args[0] == 'shadow': print(args[1] + ':' + ('!' if args[1]=='root' else '$6$hash') + ':1::::::')
|
|
else: print('127.0.0.1 STREAM example')
|
|
elif name == 'ip': print('10.203.0.1/24' if 'addr' in args else 'default via 10.0.2.2')
|
|
elif name == 'nmcli': print('connected')
|
|
else: print('fixture tool')
|
|
''')
|
|
dispatcher.chmod(0o755)
|
|
for command in ('id', 'findmnt', 'systemctl', 'getent', 'ip', 'nft',
|
|
'nmcli', 'sudo', 'hydra', 'hydra-install',
|
|
'qemu-system-x86_64', 'qemu-img', 'remote-viewer', 'foot', 'labwc'):
|
|
path = commands / command
|
|
if not path.exists():
|
|
path.symlink_to(dispatcher)
|
|
probe = self.mod.LIVE_CHECK if live else self.mod.INSTALLED_CHECK
|
|
for path in ('/etc/', '/usr/share', '/proc/cmdline', '/sys/firmware/efi', '/run/initramfs/live/LiveOS/squashfs.img'):
|
|
probe = probe.replace(path, str(root) + path)
|
|
import shutil
|
|
for command in ('python3', 'sh', 'grep', 'find', 'cut', 'readlink', 'base64'):
|
|
target = shutil.which(command)
|
|
self.assertIsNotNone(target)
|
|
(commands / command).symlink_to(target) if not (commands / command).exists() else None
|
|
env = dict(os.environ, PATH=str(commands),
|
|
SMOKE_CASE=case, SMOKE_LIVE='1' if live else '0')
|
|
result = subprocess.run(['/bin/sh', '-c', self.mod.frame_command(probe, 'probecheck')],
|
|
env=env, capture_output=True, timeout=10)
|
|
parsed = self.mod.parse_frame(result.stdout, 'probecheck')
|
|
self.assertIsNotNone(parsed, result.stderr)
|
|
return parsed[0]
|
|
|
|
def test_installed_probe_rejects_each_retained_live_artifact(self):
|
|
self.assertEqual(self.probe_fixture('clean'), 0)
|
|
for case in ('live-cmdline', 'live-account', 'live-dropin', 'live-sudoers',
|
|
'autologin', 'nopasswd', 'live-marker', 'live-greetd'):
|
|
with self.subTest(case=case):
|
|
# Give each negative case an independent clean filesystem.
|
|
import shutil
|
|
shutil.rmtree(self.root / 'guest')
|
|
self.assertNotEqual(self.probe_fixture(case), 0)
|
|
|
|
def test_installed_probe_does_not_treat_account_query_error_as_absence(self):
|
|
self.assertNotEqual(self.probe_fixture('account-lookup-error'), 0)
|
|
|
|
def test_installed_probe_rejects_unreadable_configuration(self):
|
|
self.assertNotEqual(self.probe_fixture('config-read-error'), 0)
|
|
|
|
def test_installed_probe_rejects_unreadable_boot_arguments(self):
|
|
self.assertNotEqual(self.probe_fixture('cmdline-read-error'), 0)
|
|
|
|
def test_installed_probe_rejects_each_failed_required_service(self):
|
|
self.assertEqual(self.probe_fixture('clean'), 0)
|
|
for service in ('NetworkManager', 'greetd', 'hydra-network', 'hydra-dnsmasq'):
|
|
with self.subTest(service=service):
|
|
self.assertNotEqual(self.probe_fixture(service), 0)
|
|
|
|
def test_live_probe_rejects_each_failed_required_service(self):
|
|
self.assertEqual(self.probe_fixture('clean', live=True), 0)
|
|
for service in ('NetworkManager', 'hydra-network', 'hydra-dnsmasq'):
|
|
with self.subTest(service=service):
|
|
self.assertNotEqual(self.probe_fixture(service, live=True), 0)
|
|
|
|
def tcp_probe(self, fragments):
|
|
# Execute the exact Python client embedded in the guest network probe,
|
|
# redirecting only its address to a fresh loopback socket.
|
|
import ast
|
|
import time
|
|
source = self.mod.NETWORK_CHECK.split("<<'NETWORK'\n", 1)[1].rsplit('\nNETWORK\n', 1)[0]
|
|
assignments = (node for node in ast.walk(ast.parse(source)) if isinstance(node, ast.Assign))
|
|
value = next(node.value for node in assignments
|
|
if any(isinstance(target, ast.Name) and target.id == 'client_code' for target in node.targets))
|
|
client = ast.parse(ast.literal_eval(value))
|
|
listener = socket.socket()
|
|
self.addCleanup(listener.close)
|
|
listener.bind(('127.0.0.1', 0))
|
|
listener.listen(1)
|
|
listener.settimeout(3)
|
|
host, port = listener.getsockname()
|
|
class Redirect(ast.NodeTransformer):
|
|
def visit_Tuple(self, node):
|
|
if isinstance(node.elts[0], ast.Constant) and node.elts[0].value == '10.203.0.250':
|
|
return ast.Tuple(elts=[ast.Constant(host), ast.Constant(port)], ctx=ast.Load())
|
|
return self.generic_visit(node)
|
|
client = ast.fix_missing_locations(Redirect().visit(client))
|
|
def server():
|
|
connection, _ = listener.accept()
|
|
with connection:
|
|
for fragment in fragments:
|
|
try:
|
|
connection.sendall(fragment)
|
|
except BrokenPipeError:
|
|
break
|
|
time.sleep(.05)
|
|
thread = threading.Thread(target=server)
|
|
thread.start()
|
|
result = subprocess.run(['python3', '-c', ast.unparse(client)], capture_output=True, timeout=5)
|
|
thread.join(timeout=3)
|
|
self.assertFalse(thread.is_alive())
|
|
return result
|
|
|
|
def test_tcp_probe_accepts_complete_marker_across_fragmented_reads(self):
|
|
result = self.tcp_probe([b'HYDRA_', b'LAN_TO_', b'VM_OK'])
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_tcp_probe_rejects_truncated_marker(self):
|
|
self.assertNotEqual(self.tcp_probe([b'HYDRA_LAN_TO_']).returncode, 0)
|
|
|
|
def lifecycle_fixture(self):
|
|
import json
|
|
import os
|
|
import uuid
|
|
name = 'logout-test'
|
|
runtime = self.root / 'private'
|
|
run = runtime / name
|
|
run.mkdir(parents=True, mode=0o700)
|
|
runtime.chmod(0o700)
|
|
audio = self.root / 'user-runtime'
|
|
(audio / 'pulse').mkdir(parents=True)
|
|
context = {'name': name, 'uuid': str(uuid.uuid4()), 'runtime': str(runtime),
|
|
'user_runtime': str(audio), 'uid': os.getuid()}
|
|
process = subprocess.Popen(['python3', '-c', 'import time; time.sleep(30)',
|
|
'-name', name, '-uuid', context['uuid']])
|
|
self.addCleanup(lambda: process.wait(timeout=3))
|
|
self.addCleanup(lambda: process.poll() is None and process.kill())
|
|
(run / 'qemu.pid').write_text(str(process.pid))
|
|
(run / 'identity.json').write_text(json.dumps({'name': name, 'uuid': context['uuid']}))
|
|
listeners = []
|
|
for path in (run / 'qmp.sock', run / 'spice.sock', audio / 'pipewire-0', audio / 'pulse/native'):
|
|
listener = socket.socket(socket.AF_UNIX)
|
|
listener.bind(str(path))
|
|
listener.listen(5)
|
|
listener.settimeout(.1)
|
|
listeners.append(listener)
|
|
self.addCleanup(listener.close)
|
|
stopped = threading.Event()
|
|
def qmp_server():
|
|
while not stopped.is_set():
|
|
try:
|
|
connection, _ = listeners[0].accept()
|
|
except socket.timeout:
|
|
continue
|
|
except OSError:
|
|
break
|
|
with connection, connection.makefile('rwb', buffering=0) as stream:
|
|
stream.write(b'{"QMP": {}}\n')
|
|
for line in stream:
|
|
request = json.loads(line)
|
|
replies = {'query-name': {'name': context.get('qmp_name', name)},
|
|
'query-uuid': {'UUID': context['uuid']},
|
|
'query-status': {'status': 'running'},
|
|
'query-kvm': {'enabled': True, 'present': True}}
|
|
if request['execute'] == 'query-kvm' and 'on_snapshot' in context:
|
|
context.pop('on_snapshot')()
|
|
stream.write(json.dumps({'return': replies.get(request['execute'], {}),
|
|
'id': request['id']}).encode() + b'\n')
|
|
thread = threading.Thread(target=qmp_server)
|
|
thread.start()
|
|
self.addCleanup(lambda: thread.join(timeout=3))
|
|
self.addCleanup(stopped.set)
|
|
return context, process, run
|
|
|
|
def test_logout_probe_detects_lost_private_socket(self):
|
|
context, process, run = self.lifecycle_fixture()
|
|
before = self.mod.lifecycle_snapshot(context)
|
|
self.assertEqual(self.mod.lifecycle_snapshot(context), before)
|
|
(run / 'spice.sock').unlink()
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.lifecycle_snapshot(context)
|
|
|
|
def test_logout_probe_rejects_stale_metadata_and_foreign_qmp_identity(self):
|
|
import json
|
|
context, process, run = self.lifecycle_fixture()
|
|
self.mod.lifecycle_snapshot(context)
|
|
(run / 'identity.json').write_text(json.dumps({'name': context['name'], 'uuid': 'stale'}))
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.lifecycle_snapshot(context)
|
|
(run / 'identity.json').write_text(json.dumps({'name': context['name'], 'uuid': context['uuid']}))
|
|
context['qmp_name'] = 'foreign-process'
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.lifecycle_snapshot(context)
|
|
|
|
def test_logout_probe_rejects_dead_process_and_changed_runtime_socket(self):
|
|
context, process, run = self.lifecycle_fixture()
|
|
before = self.mod.lifecycle_snapshot(context)
|
|
audio = Path(context['user_runtime']) / 'pipewire-0'
|
|
audio.unlink()
|
|
replacement = socket.socket(socket.AF_UNIX)
|
|
self.addCleanup(replacement.close)
|
|
replacement.bind(str(audio))
|
|
replacement.listen(1)
|
|
after = self.mod.lifecycle_snapshot(context)
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.verify_lifecycle_survival(before, after)
|
|
process.terminate()
|
|
process.wait(timeout=3)
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.lifecycle_snapshot(context)
|
|
|
|
def test_logout_probe_accepts_verified_same_runtime_and_rejects_process_restart(self):
|
|
context, process, run = self.lifecycle_fixture()
|
|
before = self.mod.lifecycle_snapshot(context)
|
|
self.mod.verify_lifecycle_survival(before, self.mod.lifecycle_snapshot(context))
|
|
process.terminate()
|
|
process.wait(timeout=3)
|
|
replacement = subprocess.Popen(['python3', '-c', 'import time; time.sleep(30)',
|
|
'-name', context['name'], '-uuid', context['uuid']])
|
|
self.addCleanup(lambda: replacement.wait(timeout=3))
|
|
self.addCleanup(lambda: replacement.poll() is None and replacement.kill())
|
|
(run / 'qemu.pid').write_text(str(replacement.pid))
|
|
changed = self.mod.lifecycle_snapshot(context)
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.verify_lifecycle_survival(before, changed)
|
|
|
|
def test_large_guest_probe_survives_serial_tty_canonical_line_limit(self):
|
|
import os
|
|
import pty
|
|
import termios
|
|
master, slave = pty.openpty()
|
|
self.addCleanup(os.close, master)
|
|
self.addCleanup(os.close, slave)
|
|
options = termios.tcgetattr(slave)
|
|
options[3] &= ~termios.ECHO
|
|
termios.tcsetattr(slave, termios.TCSANOW, options)
|
|
process = subprocess.Popen(['/bin/sh'], stdin=slave, stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE)
|
|
self.addCleanup(process.stdout.close)
|
|
self.addCleanup(process.stderr.close)
|
|
self.addCleanup(lambda: process.wait(timeout=3))
|
|
self.addCleanup(lambda: process.poll() is None and process.kill())
|
|
payload = 'guest-proof-' * 1200
|
|
command = self.mod.frame_command('printf %s ' + payload, 'longprobe')
|
|
os.write(master, (command + '\nexit\n').encode())
|
|
output, error = process.communicate(timeout=3)
|
|
self.assertEqual(self.mod.parse_frame(output, 'longprobe'), (0, payload.encode()), error)
|
|
|
|
def test_generated_guest_lifecycle_probe_compiles(self):
|
|
compile(self.mod.lifecycle_guest_script(), '<guest-lifecycle-probe>', 'exec')
|
|
|
|
def session_fixture(self, records):
|
|
import json
|
|
import os
|
|
from unittest.mock import patch
|
|
source = self.root / 'session-records.json'
|
|
source.write_text(json.dumps(records))
|
|
commands = self.root / 'session-fixture-bin'
|
|
commands.mkdir()
|
|
command = commands / 'loginctl'
|
|
command.write_text('''#!/usr/bin/env python3
|
|
import json, os, sys
|
|
records = json.load(open(os.environ['SMOKE_SESSION_RECORDS']))
|
|
args = sys.argv[1:]
|
|
if args[0] == 'show-user':
|
|
print(' '.join(record['Id'] for record in records if str(record['User']) == args[1]))
|
|
elif args[0] == 'show-session':
|
|
record = next((record for record in records if record['Id'] == args[1]), None)
|
|
if record is None: sys.exit(1)
|
|
for key, value in record.items(): print(key + '=' + str(value))
|
|
else: sys.exit(1)
|
|
''')
|
|
command.chmod(0o755)
|
|
environment = patch.dict(os.environ, {'PATH': str(commands) + ':' + os.environ['PATH'],
|
|
'SMOKE_SESSION_RECORDS': str(source)})
|
|
environment.start()
|
|
self.addCleanup(environment.stop)
|
|
return source
|
|
|
|
def session_record(self, ident='serial1', state='active', cls='user', leader=1234):
|
|
import os
|
|
return {'Id': ident, 'User': os.getuid(), 'State': state, 'Class': cls,
|
|
'Leader': leader, 'TTY': 'ttyS0' if cls == 'user' else ''}
|
|
|
|
def test_logout_sessions_accept_closing_scope_and_manager_with_surviving_vm(self):
|
|
context, process, run = self.lifecycle_fixture()
|
|
original = self.session_record()
|
|
context['login_session'] = original
|
|
context['before'] = self.mod.lifecycle_snapshot(context)
|
|
self.session_fixture([dict(original, State='closing'),
|
|
self.session_record('manager1', 'active', 'manager')])
|
|
self.assertTrue(self.mod.no_open_logins(context))
|
|
self.assertEqual(self.mod.capture_logout_snapshot(context), context['before'])
|
|
|
|
def test_logout_sessions_initial_login_ignores_manager_but_rejects_open_replacement(self):
|
|
import json
|
|
import os
|
|
original = self.session_record()
|
|
context = {'uid': os.getuid(), 'login_session': original}
|
|
source = self.session_fixture([original, self.session_record('manager1', 'active', 'manager')])
|
|
self.assertEqual(self.mod.require_initial_login(os.getuid(), 'serial1'), original)
|
|
for state in ('online', 'active'):
|
|
with self.subTest(state=state):
|
|
source.write_text(json.dumps([dict(original, State='closing'),
|
|
self.session_record('new-login', state)]))
|
|
self.assertFalse(self.mod.no_open_logins(context))
|
|
source.write_text(json.dumps([dict(original, State='closing', Leader=9999)]))
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.no_open_logins(context)
|
|
|
|
def test_logout_sessions_reject_login_appearing_during_snapshot(self):
|
|
import json
|
|
context, process, run = self.lifecycle_fixture()
|
|
original = self.session_record()
|
|
context['login_session'] = original
|
|
context['before'] = self.mod.lifecycle_snapshot(context)
|
|
source = self.session_fixture([dict(original, State='closing')])
|
|
context['on_snapshot'] = lambda: source.write_text(json.dumps([
|
|
dict(original, State='closing'), self.session_record('relogin')]))
|
|
self.assertTrue(self.mod.no_open_logins(context))
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.capture_logout_snapshot(context)
|
|
|
|
def test_guest_tools_accept_gentoo_remote_viewer_without_virt_viewer(self):
|
|
self.assertEqual(self.probe_fixture('clean'), 0)
|
|
self.assertEqual(self.probe_fixture('clean', live=True), 0)
|
|
|
|
|
|
def test_terminal_ux_fixture_has_complete_iso_volume_and_nonboot_directory(self):
|
|
self.assertTrue(hasattr(self.mod, 'terminal_ux_iso'), 'Installed UX ISO fixture is missing')
|
|
from hydra.media import validate_iso, MediaError
|
|
iso = self.root / 'ux.iso'
|
|
data = self.mod.terminal_ux_iso()
|
|
iso.write_bytes(data)
|
|
self.assertEqual(validate_iso(iso), iso)
|
|
self.assertEqual(len(data), 21 * 2048)
|
|
self.assertEqual(data[16 * 2048:16 * 2048 + 7], b'\x01CD001\x01')
|
|
self.assertEqual(data[17 * 2048:17 * 2048 + 7], b'\xffCD001\x01')
|
|
self.assertEqual(data[20 * 2048 + 25], 2, 'Root directory must be a directory record')
|
|
self.assertNotIn(b'EL TORITO SPECIFICATION', data, 'Fixture must not imply an OS boot')
|
|
iso.write_bytes(data[:-2048])
|
|
with self.assertRaises(MediaError):
|
|
validate_iso(iso)
|
|
|
|
def test_terminal_ux_branding_verifies_packaged_art_with_colors_and_side_data(self):
|
|
logo = '$1 @@@ @@@ @@@\n %%%%%\n\n$2 H Y D R A O S\n'
|
|
rendered = (' @@@ @@@ @@@ hydratest@hydra-os\n'
|
|
' %%%%% OS: Hydra OS x86_64\n'
|
|
' Kernel: Linux\n'
|
|
' H Y D R A O S Uptime: 1 minute\n')
|
|
actual = self.mod.verify_terminal_ux_branding(
|
|
'\x1b[36m' + rendered + '\x1b[0m', '', logo)
|
|
self.assertEqual(actual, rendered)
|
|
|
|
def test_terminal_ux_branding_rejects_incomplete_or_wrong_art_and_render_errors(self):
|
|
logo = '$1 @@@\n %%%\n %%%\n$2 H Y D R A O S\n'
|
|
rendered = ' @@@\n %%%\n %%%\n H Y D R A O S\nOS: Hydra OS x86_64\n'
|
|
for output, error in ((rendered.replace('OS: Hydra OS', 'OS: Gentoo'), ''),
|
|
(rendered.replace(' @@@\n', ''), ''),
|
|
(rendered.replace(' @@@', ' @@@@'), ''),
|
|
(rendered.replace(' @@@', ' @@@ xx'), ''),
|
|
(rendered.replace(' %%%\n', '', 1), ''),
|
|
(rendered.replace(' @@@\n %%%', ' %%%\n @@@'), ''),
|
|
(rendered.replace('H Y D R A O S', 'OTHER OS'), ''),
|
|
('H Y D R A O S\nOS: Hydra OS x86_64\n', ''),
|
|
(rendered, 'Unknown JSON key: invalid'),
|
|
(rendered + 'Memory: Error loading module\n', '')):
|
|
with self.subTest(output=output, error=error):
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.verify_terminal_ux_branding(output, error, logo)
|
|
|
|
def test_terminal_ux_branding_requires_nonblank_packaged_art_and_caption(self):
|
|
rendered = ' @@@\n H Y D R A O S\nOS: Hydra OS x86_64\n'
|
|
for logo in ('', ' \n', '$2 H Y D R A O S\n', '$1 @@@\n'):
|
|
with self.subTest(logo=logo):
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.verify_terminal_ux_branding(rendered, '', logo)
|
|
|
|
def test_terminal_ux_snapshot_checks_live_qmp_identity_and_same_process(self):
|
|
self.assertTrue(hasattr(self.mod, 'terminal_ux_snapshot'), 'Installed UX identity verification is missing')
|
|
context, process, run = self.lifecycle_fixture()
|
|
before = self.mod.terminal_ux_snapshot(context)
|
|
self.assertEqual(before['pid'], process.pid)
|
|
self.assertEqual(before['uuid'], context['uuid'])
|
|
self.assertEqual(self.mod.terminal_ux_snapshot(context), before)
|
|
context['qmp_name'] = 'foreign-process'
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.terminal_ux_snapshot(context)
|
|
context.pop('qmp_name')
|
|
process.terminate(); process.wait(timeout=3)
|
|
with self.assertRaises(self.mod.SmokeError):
|
|
self.mod.terminal_ux_snapshot(context)
|
|
|
|
def test_installed_terminal_ux_script_compiles_as_standalone_guest_probe(self):
|
|
self.assertTrue(hasattr(self.mod, 'terminal_ux_guest_script'), 'Installed terminal UX probe is missing')
|
|
compile(self.mod.terminal_ux_guest_script(), '<guest-terminal-ux>', 'exec')
|