1#!/usr/bin/env python3
2"""VM library and trust-boundary regressions, without a hypervisor."""
3import importlib.util
4import json
5import os
6import pty
7import select
8from pathlib import Path
9import socket
10import struct
11import tempfile
12import threading
13import unittest
14from unittest.mock import patch
15import sys
16from types import SimpleNamespace
17sys.path.insert(0, str(Path(__file__).parent))
18import vms
19spec = importlib.util.spec_from_file_location('vm_host', Path(__file__).with_name('dashboard-host.py'))
20host = importlib.util.module_from_spec(spec)
21spec.loader.exec_module(host)
22
23
24class VMTests(unittest.TestCase):
25 def test_nested_installers_are_separate_from_prepared_guests(self):
26 with tempfile.TemporaryDirectory() as temporary:
27 root = Path(temporary).resolve()
28 installers = root / 'Install Disks'
29 installers.mkdir()
30 (installers / 'ubuntu-amd64.ISO').write_bytes(b'installer')
31 (installers / 'ubuntu-arm64.iso').write_bytes(b'installer')
32 (installers / 'not-a-preset.qcow2').write_bytes(b'disk')
33 (root / 'restore.ipsw').write_bytes(b'restore')
34 (installers / 'escaped.iso').symlink_to(root / 'restore.ipsw')
35 presets = root / 'Presets'
36 presets.mkdir()
37 (presets / 'ignored.iso').write_bytes(b'not an installer')
38 with patch.multiple(vms, IMAGES=root, UPLOADS=installers, PRESETS=presets):
39 library = vms.library()
40 self.assertEqual(len(library['installers']), 2)
41 self.assertEqual(library['presets'], [])
42 self.assertEqual({item['arch'] for item in library['installers']}, {'x86_64', 'aarch64'})
43 self.assertTrue(all(item['volume'].startswith('media:Install Disks/') for item in library['installers']))
44
45 def test_image_ids_and_parent_symlinks_cannot_escape(self):
46 for identity in ['../secret.iso', 'media:Install Disks/../../secret.iso', '/etc/shadow', 'upload:/secret.iso']:
47 with self.subTest(identity=identity), self.assertRaises(ValueError):
48 vms.validate_image_id(identity)
49 with tempfile.TemporaryDirectory() as temporary:
50 root = Path(temporary).resolve()
51 (root / 'outside').mkdir()
52 (root / 'outside' / 'secret.iso').write_bytes(b'secret')
53 (root / 'inside').symlink_to(root / 'outside', target_is_directory=True)
54 with self.assertRaises(OSError):
55 vms.open_regular(root / 'inside' / 'secret.iso')
56
57 def test_modern_create_rejects_incomplete_or_arbitrary_hardware(self):
58 payload = {'name': 'ubuntu', 'owner': 'fixture-user', 'username': 'fixture', 'description': '', 'image': 'media:Install Disks/ubuntu.iso',
59 'mode': 'iso', 'firmware': 'uefi', 'platform': 'linux', 'vcpus': 2,
60 'memory': 2**30, 'disk': 20 * 2**30, 'autostart': False, 'start': False}
61 with patch.object(vms, 'node', return_value={'cpus': 8, 'memory': 16 * 2**30}):
62 vms.validate('create', payload)
63 for changes in [{'mode': 'xml'}, {'firmware': '/etc/shadow'}, {'vcpus': True}, {'image': '../escape'}, {'argv': ['sh']}]:
64 with self.subTest(changes=changes), self.assertRaises(ValueError):
65 vms.validate('create', {**payload, **changes})
66 with self.assertRaises(ValueError):
67 vms.validate('create', {'name': 'ubuntu'})
68
69 def test_known_installers_choose_their_tested_hardware(self):
70 linux = {'platform': 'linux', 'firmware': 'uefi'}
71 windows = {'platform': 'windows', 'firmware': 'uefi'}
72 cases = {
73 'Windows XP Professional SP3 x86.iso': 'windows-legacy-ide',
74 'Windows Vista SP2 x64.iso': 'windows-legacy-ide',
75 'Windows 7 Professional.ISO': 'windows-legacy-ide',
76 'Windows 8 RTM x64.iso': 'windows-q35-bios',
77 'Windows 10 22H2 English x64.iso': 'windows-q35-uefi',
78 'Windows 11 26H2 English x64.iso': 'windows-q35-secure',
79 'Fedora-Workstation-Live-44-1.7.aarch64.iso': 'linux-aarch64-virtio',
80 }
81 for name, expected in cases.items():
82 with self.subTest(name=name):
83 spec = windows if name.startswith('Windows') else linux
84 self.assertEqual(vms.installer_profile(Path(name), spec), expected)
85 self.assertEqual(vms.profile('linux-aarch64-virtio')['video'], 'virtio')
86 self.assertEqual(vms.profile('linux-aarch64-virtio')['keyboard'], 'usb')
87 self.assertEqual(vms.profile('linux-aarch64-virtio')['serial'], 'ttyAMA0')
88 self.assertEqual(vms.profile('windows-q35-secure')['cpu_mode'], 'host-passthrough')
89 self.assertEqual(vms.profile('windows-q35-secure')['clock'], 'utc')
90 with self.assertRaisesRegex(ValueError, 'Windows ARM'):
91 vms.installer_profile(Path('Windows 11 26H2 English arm64.iso'), windows)
92
93 def test_existing_presets_receive_a_safe_profile(self):
94 with tempfile.TemporaryDirectory() as temporary:
95 presets = Path(temporary).resolve()
96 directory = presets / 'ubuntu'
97 directory.mkdir()
98 (directory / 'preset.json').write_text(json.dumps({
99 'id': 'ubuntu', 'os': 'Ubuntu', 'description': '', 'firmware': 'uefi',
100 'platform': 'linux', 'createdAt': 0, 'recommended': {'vcpus': 2, 'memory': 1, 'disk': 1},
101 }))
102 (directory / 'disk.qcow2').write_bytes(b'fixture')
103 details = {'format': 'qcow2', 'virtual-size': 123, 'format-specific': {'data': {}}}
104 with patch.object(vms, 'PRESETS', presets), patch.object(vms, 'command', return_value=json.dumps(details)):
105 preset = vms.read_preset('ubuntu')
106 self.assertEqual(preset['hardwareProfile'], 'linux-x86-virtio')
107 self.assertEqual(preset['arch'], 'x86_64')
108
109 def test_windows_seed_has_only_validated_clone_identity(self):
110 with tempfile.TemporaryDirectory() as temporary:
111 directory = Path(temporary)
112 calls = []
113 with patch.object(vms, 'command', side_effect=lambda *args, **kwargs: calls.append(args) or ''), \
114 patch.object(vms.secrets, 'token_urlsafe', return_value='generated-password'), \
115 patch.object(vms.secrets, 'token_hex', return_value='1234abcd'):
116 vms.windows_seed(directory, 'snow')
117 self.assertEqual(json.loads((directory / 'access.json').read_text()), {
118 'username': 'snow', 'password': 'generated-password', 'hostname': 'sgvm-1234abcd',
119 })
120 self.assertEqual(oct((directory / 'access.json').stat().st_mode & 0o777), '0o600')
121 self.assertEqual(calls[0][0:7], ('genisoimage', '-quiet', '-output', str(directory / 'seed.iso'), '-volid', 'SNOWGLOBE', '-joliet'))
122 self.assertEqual(Path(calls[0][-1]).name, 'SNOWGLOB.INI')
123 with self.assertRaisesRegex(ValueError, 'Windows'):
124 vms.windows_seed(directory, 'bad\\nname')
125
126 def test_nixos_seed_uses_mutable_gdm_path_and_native_sshd(self):
127 with tempfile.TemporaryDirectory() as temporary:
128 directory = Path(temporary)
129 captured = {}
130 def image(*args, **kwargs):
131 captured['config'] = json.loads(Path(args[-2]).read_text().removeprefix('#cloud-config\n'))
132 return ''
133 with patch.object(vms, 'command', side_effect=image), \
134 patch.object(vms.subprocess, 'run', return_value=SimpleNamespace(stdout='$6$hash\n')), \
135 patch.object(vms.secrets, 'token_urlsafe', return_value='generated-password'), \
136 patch.object(vms.secrets, 'token_hex', side_effect=['1234abcd', 'instance']):
137 vms.linux_seed(directory, 'snow', 'NixOS', 'ttyAMA0')
138 config = captured['config']
139 self.assertEqual(config['bootcmd'][0], ['rm', '-f', '/etc/gdm/custom.conf'])
140 self.assertIn('/run/current-system/sw/bin/getent', config['bootcmd'][1][-1])
141 self.assertEqual(config['write_files'], [{
142 'path': '/etc/gdm/custom.conf', 'content': '[daemon]\nAutomaticLoginEnable=true\nAutomaticLogin=snow\n',
143 }, {
144 'path': '/var/lib/snowglobe/hostname',
145 'content': 'snowglobe-vm-1234abcd\n',
146 }])
147 self.assertEqual(config['runcmd'][0][-1], 'sshd')
148 self.assertEqual(config['runcmd'][1][-1], 'serial-getty@ttyAMA0.service')
149
150 def test_console_cannot_choose_an_arbitrary_network_target(self):
151 xml = "<domain><uuid>fixture-vm</uuid><devices><graphics type='vnc' listen='127.0.0.1' port='5901'/></devices></domain>"
152 def virsh(*args):
153 return 'running' if args[0] == 'domstate' else xml
154 with patch.object(vms, 'virsh', side_effect=virsh):
155 self.assertEqual(vms.console_target('ubuntu'), {'port': 5901, 'uuid': 'fixture-vm'})
156 for address in ['0.0.0.0', '192.168.0.1', 'localhost']:
157 xml = xml.replace('127.0.0.1', address)
158 with self.assertRaises(ValueError):
159 vms.console_target('ubuntu')
160 xml = xml.replace(address, '127.0.0.1')
161 xml = xml.replace('5901', '22')
162 with self.assertRaises(ValueError):
163 vms.console_target('ubuntu')
164
165 def test_running_guests_cannot_be_published_as_presets(self):
166 with patch.object(vms, 'virsh', return_value='running'):
167 with self.assertRaisesRegex(ValueError, 'Shut down'):
168 vms.save_preset({'name': 'ubuntu'})
169
170 def test_screen_waits_for_startup_without_following_a_replacement_vm(self):
171 with socket.socket() as reservation:
172 reservation.bind(('127.0.0.1', 0))
173 port = reservation.getsockname()[1]
174 with socket.socket() as display:
175 retry, listening = threading.Event(), threading.Event()
176 errors = []
177 class StartingHost:
178 calls = 0
179 def handle(self, request):
180 self.calls += 1
181 if self.calls == 2:
182 retry.set()
183 if not listening.wait(2):
184 raise TimeoutError('Display did not start')
185 return {'port': port, 'uuid': 'original-vm'}
186 target = StartingHost()
187 client, server = socket.socketpair()
188 client.settimeout(2)
189 def run():
190 with server:
191 try:
192 host.stream_console(server, target, {'operation': 'vm.console'})
193 except Exception as error:
194 errors.append(error)
195 worker = threading.Thread(target=run)
196 worker.start()
197 try:
198 self.assertTrue(retry.wait(2))
199 display.bind(('127.0.0.1', port))
200 display.listen()
201 display.settimeout(2)
202 listening.set()
203 with display.accept()[0] as guest:
204 length = struct.unpack('!I', client.recv(4))[0]
205 self.assertEqual(json.loads(client.recv(length)), {'value': None})
206 guest.sendall(b'RFB 003.008\n')
207 self.assertEqual(client.recv(12), b'RFB 003.008\n')
208 finally:
209 listening.set()
210 client.close()
211 worker.join(2)
212 self.assertFalse(worker.is_alive())
213 self.assertEqual(errors, [])
214 with patch.object(host.socket, 'create_connection', side_effect=ConnectionRefusedError) as connect:
215 target = StartingHost()
216 with patch.object(target, 'handle', side_effect=[{'port': port, 'uuid': 'original-vm'}, {'port': port, 'uuid': 'replacement-vm'}]):
217 with self.assertRaisesRegex(host.Rejected, 'VM changed'):
218 host.stream_console(None, target, {'operation': 'vm.console'})
219 self.assertEqual(connect.call_count, 1)
220
221 def test_serial_target_cannot_choose_a_host_file(self):
222 xml = "<domain><devices><console type='pty'><source path='/dev/pts/3'/><target type='serial'/></console></devices></domain>"
223 def virsh(*args):
224 return 'running' if args[0] == 'domstate' else xml
225 with patch.object(sys, 'argv', ['vms.py', 'serial', json.dumps({'name': 'ubuntu'})]), patch.object(vms, 'virsh', side_effect=virsh):
226 self.assertEqual(vms.main(), {'path': '/dev/pts/3'})
227 for path in ['/etc/shadow', '/dev/null', '/dev/pts/../tty', '/dev/pts/3/link']:
228 xml = xml.replace('/dev/pts/3', path)
229 with self.assertRaises(ValueError):
230 vms.main()
231 xml = xml.replace(path, '/dev/pts/3')
232
233 def test_serial_transport_preserves_control_c_in_both_directions(self):
234 master, slave = pty.openpty()
235 client, server = socket.socketpair()
236 client.settimeout(2)
237 errors = []
238 class SerialHost:
239 def handle(self, request):
240 return {'path': os.ttyname(slave)}
241 def run():
242 with server:
243 try:
244 host.stream_console(server, SerialHost(), {'operation': 'vm.serial'})
245 except Exception as error:
246 errors.append(error)
247 worker = threading.Thread(target=run)
248 worker.start()
249 try:
250 length = struct.unpack('!I', client.recv(4))[0]
251 self.assertEqual(json.loads(client.recv(length)), {'value': None})
252 other_client, other_server = socket.socketpair()
253 with other_client, other_server, self.assertRaisesRegex(host.Rejected, 'already open'):
254 host.stream_console(other_server, SerialHost(), {'operation': 'vm.serial'})
255 client.sendall(b'\x03')
256 self.assertTrue(select.select([master], [], [], 2)[0])
257 self.assertEqual(os.read(master, 100), b'\x03')
258 os.write(master, b'guest prompt> ')
259 self.assertEqual(client.recv(100), b'guest prompt> ')
260 finally:
261 client.close()
262 worker.join(2)
263 os.close(master)
264 os.close(slave)
265 self.assertFalse(worker.is_alive())
266 self.assertEqual(errors, [])
267
268 def test_upload_stream_is_bounded_atomic_and_cleans_interruption(self):
269 def response(sock):
270 length = struct.unpack('!I', sock.recv(4))[0]
271 data = bytearray()
272 while len(data) < length:
273 data.extend(sock.recv(length - len(data)))
274 return json.loads(data)
275 with tempfile.TemporaryDirectory() as temporary, patch.object(vms, 'UPLOADS', Path(temporary).resolve()):
276 for complete in [False, True]:
277 client, server = socket.socketpair()
278 errors = []
279 def run():
280 with server:
281 try:
282 host.stream_upload(server, {'operation': 'vm.upload', 'payload': {'volume': 'ubuntu.iso', 'size': 32768}})
283 except Exception as error:
284 errors.append(error)
285 worker = threading.Thread(target=run)
286 worker.start()
287 with client:
288 self.assertEqual(response(client), {'value': None})
289 client.sendall(b'I' * (32768 if complete else 100))
290 client.shutdown(socket.SHUT_WR)
291 if complete:
292 self.assertEqual(response(client)['value']['volume'], 'upload:ubuntu.iso')
293 worker.join(2)
294 self.assertFalse(worker.is_alive())
295 if complete:
296 self.assertEqual((Path(temporary).resolve() / 'ubuntu.iso').read_bytes(), b'I' * 32768)
297 else:
298 self.assertTrue(errors)
299 self.assertEqual(list(Path(temporary).resolve().iterdir()), [])
300 with self.assertRaises(host.Rejected):
301 host.stream_upload(None, {'operation': 'vm.upload', 'payload': {'volume': 'ubuntu.iso', 'size': 32768}})
302 for filename in ['../ubuntu.iso', '/etc/shadow.iso', 'ubuntu.img']:
303 with self.assertRaises(ValueError):
304 vms.validate('upload', {'volume': filename, 'size': 32768})
305
306
307if __name__ == '__main__':
308 unittest.main()