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