144 lines
6.2 KiB
Python
144 lines
6.2 KiB
Python
"""Regression tests for disposable-VM isolation and trustworthy serial evidence."""
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import socket
|
|
import subprocess
|
|
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_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 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)
|