123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830 |
- # pylint: disable=protected-access
- #
- # The Qubes OS Project, https://www.qubes-os.org/
- #
- # Copyright (C) 2014-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
- # Copyright (C) 2014-2015 Wojtek Porczyk <woju@invisiblethingslab.com>
- #
- # This library is free software; you can redistribute it and/or
- # modify it under the terms of the GNU Lesser General Public
- # License as published by the Free Software Foundation; either
- # version 2.1 of the License, or (at your option) any later version.
- #
- # This library is distributed in the hope that it will be useful,
- # but WITHOUT ANY WARRANTY; without even the implied warranty of
- # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- # Lesser General Public License for more details.
- #
- # You should have received a copy of the GNU Lesser General Public
- # License along with this library; if not, see <https://www.gnu.org/licenses/>.
- #
- import base64
- import os
- import unittest
- import uuid
- import datetime
- import lxml.etree
- import unittest.mock
- import qubes
- import qubes.exc
- import qubes.config
- import qubes.vm
- import qubes.vm.qubesvm
- import qubes.tests
- import qubes.tests.vm
- class TestApp(object):
- labels = {1: qubes.Label(1, '0xcc0000', 'red')}
- def __init__(self):
- self.domains = {}
- class TestProp(object):
- # pylint: disable=too-few-public-methods
- __name__ = 'testprop'
- class TestDeviceCollection(object):
- def __init__(self):
- self._list = []
- def persistent(self):
- return self._list
- class TestQubesDB(object):
- def __init__(self):
- self.data = {}
- def write(self, path, value):
- self.data[path] = value
- def rm(self, path):
- if path.endswith('/'):
- for key in [x for x in self.data if x.startswith(path)]:
- del self.data[key]
- else:
- self.data.pop(path, None)
- class TestVM(object):
- # pylint: disable=too-few-public-methods
- app = TestApp()
- def __init__(self, **kwargs):
- self.running = False
- self.installed_by_rpm = False
- for k, v in kwargs.items():
- setattr(self, k, v)
- self.devices = {'pci': TestDeviceCollection()}
- def is_running(self):
- return self.running
- class TC_00_setters(qubes.tests.QubesTestCase):
- def setUp(self):
- super().setUp()
- self.vm = TestVM()
- self.prop = TestProp()
- def test_000_setter_qid(self):
- self.assertEqual(
- qubes.vm._setter_qid(self.vm, self.prop, 5), 5)
- def test_001_setter_qid_lt_0(self):
- with self.assertRaises(ValueError):
- qubes.vm._setter_qid(self.vm, self.prop, -1)
- def test_002_setter_qid_gt_max(self):
- with self.assertRaises(ValueError):
- qubes.vm._setter_qid(self.vm,
- self.prop, qubes.config.max_qid + 5)
- @unittest.skip('test not implemented')
- def test_020_setter_kernel(self):
- pass
- def test_030_setter_label_object(self):
- label = TestApp.labels[1]
- self.assertIs(label,
- qubes.vm.setter_label(self.vm, self.prop, label))
- def test_031_setter_label_getitem(self):
- label = TestApp.labels[1]
- self.assertIs(label,
- qubes.vm.setter_label(self.vm, self.prop, 'label-1'))
- # there is no check for self.app.get_label()
- def test_040_setter_virt_mode(self):
- self.assertEqual(
- qubes.vm.qubesvm._setter_virt_mode(self.vm, self.prop, 'hvm'),
- 'hvm')
- self.assertEqual(
- qubes.vm.qubesvm._setter_virt_mode(self.vm, self.prop, 'HVM'),
- 'hvm')
- self.assertEqual(
- qubes.vm.qubesvm._setter_virt_mode(self.vm, self.prop, 'PV'),
- 'pv')
- self.assertEqual(
- qubes.vm.qubesvm._setter_virt_mode(self.vm, self.prop, 'pvh'),
- 'pvh')
- self.vm.devices['pci']._list.append(object())
- with self.assertRaises(ValueError):
- qubes.vm.qubesvm._setter_virt_mode(self.vm, self.prop, 'pvh')
- with self.assertRaises(ValueError):
- qubes.vm.qubesvm._setter_virt_mode(self.vm, self.prop, 'True')
- class QubesVMTestsMixin(object):
- property_no_default = object()
- def setUp(self):
- super(QubesVMTestsMixin, self).setUp()
- self.app = qubes.tests.vm.TestApp()
- self.app.vmm.offline_mode = True
- # when full test run is called, extensions are loaded by earlier
- # tests, but if just this test class is run, load them manually here,
- # to have the same behaviour
- qubes.ext.get_extensions()
- def tearDown(self):
- try:
- self.app.domains.close()
- except AttributeError:
- pass
- super(QubesVMTestsMixin, self).tearDown()
- def get_vm(self, name='test', cls=qubes.vm.qubesvm.QubesVM, **kwargs):
- vm = cls(self.app, None,
- qid=kwargs.pop('qid', 1), name=qubes.tests.VMPREFIX + name,
- **kwargs)
- self.app.domains[vm.qid] = vm
- self.app.domains[vm.uuid] = vm
- self.app.domains[vm] = vm
- self.addCleanup(vm.close)
- return vm
- def assertPropertyValue(self, vm, prop_name, set_value, expected_value,
- expected_xml_content=None):
- # FIXME: any better exception list? or maybe all of that should be a
- # single exception?
- with self.assertNotRaises((ValueError, TypeError, KeyError)):
- setattr(vm, prop_name, set_value)
- self.assertEqual(getattr(vm, prop_name), expected_value)
- if expected_xml_content is not None:
- xml = vm.__xml__()
- prop_xml = xml.xpath(
- './properties/property[@name=\'{}\']'.format(prop_name))
- self.assertEqual(len(prop_xml), 1, "Property not found in XML")
- self.assertEqual(prop_xml[0].text, expected_xml_content)
- def assertPropertyInvalidValue(self, vm, prop_name, set_value):
- orig_value_set = True
- orig_value = None
- try:
- orig_value = getattr(vm, prop_name)
- except AttributeError:
- orig_value_set = False
- # FIXME: any better exception list? or maybe all of that should be a
- # single exception?
- with self.assertRaises((ValueError, TypeError, KeyError)):
- setattr(vm, prop_name, set_value)
- if orig_value_set:
- self.assertEqual(getattr(vm, prop_name), orig_value)
- else:
- with self.assertRaises(AttributeError):
- getattr(vm, prop_name)
- def assertPropertyDefaultValue(self, vm, prop_name,
- expected_default=property_no_default):
- if expected_default is self.property_no_default:
- with self.assertRaises(AttributeError):
- getattr(vm, prop_name)
- else:
- with self.assertNotRaises(AttributeError):
- self.assertEqual(getattr(vm, prop_name), expected_default)
- xml = vm.__xml__()
- prop_xml = xml.xpath(
- './properties/property[@name=\'{}\']'.format(prop_name))
- self.assertEqual(len(prop_xml), 0, "Property still found in XML")
- def _test_generic_bool_property(self, vm, prop_name, default=False):
- self.assertPropertyDefaultValue(vm, prop_name, default)
- self.assertPropertyValue(vm, prop_name, False, False, 'False')
- self.assertPropertyValue(vm, prop_name, True, True, 'True')
- delattr(vm, prop_name)
- self.assertPropertyDefaultValue(vm, prop_name, default)
- self.assertPropertyValue(vm, prop_name, 'True', True, 'True')
- self.assertPropertyValue(vm, prop_name, 'False', False, 'False')
- self.assertPropertyInvalidValue(vm, prop_name, 'xxx')
- self.assertPropertyValue(vm, prop_name, 123, True)
- self.assertPropertyInvalidValue(vm, prop_name, '')
- class TC_90_QubesVM(QubesVMTestsMixin, qubes.tests.QubesTestCase):
- def test_000_init(self):
- self.get_vm()
- def test_001_init_no_qid_or_name(self):
- with self.assertRaises(AssertionError):
- qubes.vm.qubesvm.QubesVM(self.app, None,
- name=qubes.tests.VMPREFIX + 'test')
- with self.assertRaises(AssertionError):
- qubes.vm.qubesvm.QubesVM(self.app, None,
- qid=1)
- def test_003_init_fire_domain_init(self):
- class TestVM2(qubes.vm.qubesvm.QubesVM):
- event_fired = False
- @qubes.events.handler('domain-init')
- def on_domain_init(self, event): # pylint: disable=unused-argument
- self.__class__.event_fired = True
- TestVM2(self.app, None, qid=1, name=qubes.tests.VMPREFIX + 'test')
- self.assertTrue(TestVM2.event_fired)
- def test_004_uuid_autogen(self):
- vm = self.get_vm()
- self.assertTrue(hasattr(vm, 'uuid'))
- def test_100_qid(self):
- vm = self.get_vm()
- self.assertIsInstance(vm.qid, int)
- with self.assertRaises(AttributeError):
- vm.qid = 2
- def test_110_name(self):
- vm = self.get_vm()
- self.assertIsInstance(vm.name, str)
- def test_120_uuid(self):
- my_uuid = uuid.uuid4()
- vm = self.get_vm(uuid=my_uuid)
- self.assertIsInstance(vm.uuid, uuid.UUID)
- self.assertIs(vm.uuid, my_uuid)
- with self.assertRaises(AttributeError):
- vm.uuid = uuid.uuid4()
- @unittest.skip('TODO: how to not fail on making an icon symlink here?')
- def test_130_label(self):
- vm = self.get_vm()
- self.assertPropertyDefaultValue(vm, 'label')
- self.assertPropertyValue(vm, 'label', self.app.labels[1],
- self.app.labels[1], 'label-1')
- del vm.label
- self.assertPropertyDefaultValue(vm, 'label')
- self.assertPropertyValue(vm, 'label', 'red',
- self.app.labels[1], 'label-1')
- self.assertPropertyValue(vm, 'label', 'label-1',
- self.app.labels[1], 'label-1')
- def test_131_label_invalid(self):
- vm = self.get_vm()
- self.assertPropertyInvalidValue(vm, 'label', 'invalid')
- self.assertPropertyInvalidValue(vm, 'label', 123)
- def test_160_memory(self):
- vm = self.get_vm()
- self.assertPropertyDefaultValue(vm, 'memory', 400)
- self.assertPropertyValue(vm, 'memory', 500, 500, '500')
- del vm.memory
- self.assertPropertyDefaultValue(vm, 'memory', 400)
- self.assertPropertyValue(vm, 'memory', '500', 500, '500')
- def test_161_memory_invalid(self):
- vm = self.get_vm()
- self.assertPropertyInvalidValue(vm, 'memory', -100)
- self.assertPropertyInvalidValue(vm, 'memory', '-100')
- self.assertPropertyInvalidValue(vm, 'memory', '')
- # TODO: higher than maxmem
- # TODO: human readable setter (500M, 4G)?
- def test_170_maxmem(self):
- vm = self.get_vm()
- self.assertPropertyDefaultValue(vm, 'maxmem',
- self.app.host.memory_total / 1024 / 2)
- self.assertPropertyValue(vm, 'maxmem', 500, 500, '500')
- del vm.maxmem
- self.assertPropertyDefaultValue(vm, 'maxmem',
- self.app.host.memory_total / 1024 / 2)
- self.assertPropertyValue(vm, 'maxmem', '500', 500, '500')
- def test_171_maxmem_invalid(self):
- vm = self.get_vm()
- self.assertPropertyInvalidValue(vm, 'maxmem', -100)
- self.assertPropertyInvalidValue(vm, 'maxmem', '-100')
- self.assertPropertyInvalidValue(vm, 'maxmem', '')
- # TODO: lower than memory
- # TODO: human readable setter (500M, 4G)?
- def test_190_vcpus(self):
- vm = self.get_vm()
- self.assertPropertyDefaultValue(vm, 'vcpus', 2)
- self.assertPropertyValue(vm, 'vcpus', 3, 3, '3')
- del vm.vcpus
- self.assertPropertyDefaultValue(vm, 'vcpus', 2)
- self.assertPropertyValue(vm, 'vcpus', '3', 3, '3')
- def test_191_vcpus_invalid(self):
- vm = self.get_vm()
- self.assertPropertyInvalidValue(vm, 'vcpus', 0)
- self.assertPropertyInvalidValue(vm, 'vcpus', -2)
- self.assertPropertyInvalidValue(vm, 'vcpus', '-2')
- self.assertPropertyInvalidValue(vm, 'vcpus', '')
- def test_200_debug(self):
- vm = self.get_vm()
- self._test_generic_bool_property(vm, 'debug', False)
- def test_210_installed_by_rpm(self):
- vm = self.get_vm()
- self._test_generic_bool_property(vm, 'installed_by_rpm', False)
- def test_220_include_in_backups(self):
- vm = self.get_vm()
- self._test_generic_bool_property(vm, 'include_in_backups', True)
- @qubes.tests.skipUnlessDom0
- def test_250_kernel(self):
- kernels = os.listdir(os.path.join(
- qubes.config.qubes_base_dir,
- qubes.config.system_path['qubes_kernels_base_dir']))
- if not len(kernels):
- self.skipTest('Needs at least one kernel installed')
- self.app.default_kernel = kernels[0]
- vm = self.get_vm()
- self.assertPropertyDefaultValue(vm, 'kernel', kernels[0])
- self.assertPropertyValue(vm, 'kernel', kernels[-1], kernels[-1],
- kernels[-1])
- del vm.kernel
- self.assertPropertyDefaultValue(vm, 'kernel', kernels[0])
- @qubes.tests.skipUnlessDom0
- def test_251_kernel_invalid(self):
- vm = self.get_vm()
- self.assertPropertyInvalidValue(vm, 'kernel', 123)
- self.assertPropertyInvalidValue(vm, 'kernel', 'invalid')
- def test_252_kernel_empty(self):
- vm = self.get_vm()
- self.assertPropertyValue(vm, 'kernel', '', '', '')
- self.assertPropertyValue(vm, 'kernel', None, '', '')
- def test_260_kernelopts(self):
- vm = self.get_vm()
- self.assertPropertyDefaultValue(vm, 'kernelopts',
- qubes.config.defaults['kernelopts'])
- self.assertPropertyValue(vm, 'kernelopts', 'some options',
- 'some options', 'some options')
- del vm.kernelopts
- self.assertPropertyDefaultValue(vm, 'kernelopts',
- qubes.config.defaults['kernelopts'])
- self.assertPropertyValue(vm, 'kernelopts', '',
- '', '')
- # TODO?
- # self.assertPropertyInvalidValue(vm, 'kernelopts', None),
- @unittest.skip('test not implemented')
- def test_261_kernelopts_pcidevs(self):
- vm = self.get_vm()
- # how to do that here? use dummy DeviceManager/DeviceCollection?
- # Disable events?
- vm.devices['pci'].attach('something')
- self.assertPropertyDefaultValue(vm, 'kernelopts',
- qubes.config.defaults['kernelopts_pcidevs'])
- def test_270_qrexec_timeout(self):
- vm = self.get_vm()
- self.assertPropertyDefaultValue(vm, 'qrexec_timeout', 60)
- self.assertPropertyValue(vm, 'qrexec_timeout', 3, 3, '3')
- del vm.qrexec_timeout
- self.assertPropertyDefaultValue(vm, 'qrexec_timeout', 60)
- self.assertPropertyValue(vm, 'qrexec_timeout', '3', 3, '3')
- def test_271_qrexec_timeout_invalid(self):
- vm = self.get_vm()
- self.assertPropertyInvalidValue(vm, 'qrexec_timeout', -2)
- self.assertPropertyInvalidValue(vm, 'qrexec_timeout', '-2')
- self.assertPropertyInvalidValue(vm, 'qrexec_timeout', '')
- def test_280_autostart(self):
- vm = self.get_vm()
- # FIXME any better idea to not involve systemctl call at this stage?
- vm.events_enabled = False
- self._test_generic_bool_property(vm, 'autostart', False)
- @qubes.tests.skipUnlessDom0
- def test_281_autostart_systemd(self):
- vm = self.get_vm()
- self.assertFalse(os.path.exists(
- '/etc/systemd/system/multi-user.target.wants/'
- 'qubes-vm@{}.service'.format(vm.name)),
- "systemd service enabled before setting autostart")
- vm.autostart = True
- self.assertTrue(os.path.exists(
- '/etc/systemd/system/multi-user.target.wants/'
- 'qubes-vm@{}.service'.format(vm.name)),
- "systemd service not enabled by autostart=True")
- vm.autostart = False
- self.assertFalse(os.path.exists(
- '/etc/systemd/system/multi-user.target.wants/'
- 'qubes-vm@{}.service'.format(vm.name)),
- "systemd service not disabled by autostart=False")
- vm.autostart = True
- del vm.autostart
- self.assertFalse(os.path.exists(
- '/etc/systemd/system/multi-user.target.wants/'
- 'qubes-vm@{}.service'.format(vm.name)),
- "systemd service not disabled by resetting autostart")
- @unittest.skip('TODO')
- def test_320_seamless_gui_mode(self):
- vm = self.get_vm()
- self._test_generic_bool_property(vm, 'seamless_gui_mode')
- # TODO: reject setting to True when guiagent_installed is false
- def test_330_mac(self):
- vm = self.get_vm()
- # TODO: calculate proper default here
- default_mac = vm.mac
- self.assertIsNotNone(default_mac)
- self.assertPropertyDefaultValue(vm, 'mac', default_mac)
- self.assertPropertyValue(vm, 'mac', '00:11:22:33:44:55',
- '00:11:22:33:44:55', '00:11:22:33:44:55')
- del vm.mac
- self.assertPropertyDefaultValue(vm, 'mac', default_mac)
- def test_331_mac_invalid(self):
- vm = self.get_vm()
- self.assertPropertyInvalidValue(vm, 'mac', 123)
- self.assertPropertyInvalidValue(vm, 'mac', 'invalid')
- self.assertPropertyInvalidValue(vm, 'mac', '00:11:22:33:44:55:66')
- def test_340_default_user(self):
- vm = self.get_vm()
- self.assertPropertyDefaultValue(vm, 'default_user', 'user')
- self.assertPropertyValue(vm, 'default_user', 'someuser', 'someuser',
- 'someuser')
- del vm.default_user
- self.assertPropertyDefaultValue(vm, 'default_user', 'user')
- self.assertPropertyValue(vm, 'default_user', 123, '123', '123')
- # TODO: check propagation for template-based VMs
- @unittest.skip('TODO')
- def test_350_timezone(self):
- vm = self.get_vm()
- self.assertPropertyDefaultValue(vm, 'timezone', 'localtime')
- self.assertPropertyValue(vm, 'timezone', 0, 0, '0')
- del vm.timezone
- self.assertPropertyDefaultValue(vm, 'timezone', 'localtime')
- self.assertPropertyValue(vm, 'timezone', '0', 0, '0')
- self.assertPropertyValue(vm, 'timezone', -3600, -3600, '-3600')
- self.assertPropertyValue(vm, 'timezone', 7200, 7200, '7200')
- @unittest.skip('TODO')
- def test_350_timezone_invalid(self):
- vm = self.get_vm()
- self.assertPropertyInvalidValue(vm, 'timezone', 'xxx')
- @unittest.skip('TODO')
- def test_360_drive(self):
- vm = self.get_vm()
- self.assertPropertyDefaultValue(vm, 'drive', None)
- # self.execute_tests('drive', [
- # ('hd:dom0:/tmp/drive.img', 'hd:dom0:/tmp/drive.img', True),
- # ('hd:/tmp/drive.img', 'hd:dom0:/tmp/drive.img', True),
- # ('cdrom:dom0:/tmp/drive.img', 'cdrom:dom0:/tmp/drive.img', True),
- # ('cdrom:/tmp/drive.img', 'cdrom:dom0:/tmp/drive.img', True),
- # ('/tmp/drive.img', 'cdrom:dom0:/tmp/drive.img', True),
- # ('hd:drive.img', '', False),
- # ('drive.img', '', False),
- # ])
- def test_400_backup_timestamp(self):
- vm = self.get_vm()
- timestamp = datetime.datetime(2016, 1, 1, 12, 14, 2)
- timestamp_str = timestamp.strftime('%s')
- self.assertPropertyDefaultValue(vm, 'backup_timestamp', None)
- self.assertPropertyValue(vm, 'backup_timestamp', timestamp,
- timestamp, timestamp_str)
- del vm.backup_timestamp
- self.assertPropertyDefaultValue(vm, 'backup_timestamp', None)
- self.assertPropertyValue(vm, 'backup_timestamp', timestamp_str,
- timestamp)
- def test_401_backup_timestamp_invalid(self):
- vm = self.get_vm()
- self.assertPropertyInvalidValue(vm, 'backup_timestamp', 'xxx')
- self.assertPropertyInvalidValue(vm, 'backup_timestamp', None)
- def test_500_property_migrate_virt_mode(self):
- xml_template = '''
- <domain class="QubesVM" id="domain-1">
- <properties>
- <property name="qid">1</property>
- <property name="name">testvm</property>
- <property name="label" ref="label-1" />
- <property name="hvm">{hvm_value}</property>
- </properties>
- </domain>
- '''
- xml = lxml.etree.XML(xml_template.format(hvm_value='True'))
- vm = qubes.vm.qubesvm.QubesVM(self.app, xml)
- self.assertEqual(vm.virt_mode, 'hvm')
- with self.assertRaises(AttributeError):
- vm.hvm
- xml = lxml.etree.XML(xml_template.format(hvm_value='False'))
- vm = qubes.vm.qubesvm.QubesVM(self.app, xml)
- self.assertEqual(vm.virt_mode, 'pv')
- with self.assertRaises(AttributeError):
- vm.hvm
- def test_600_libvirt_xml_pv(self):
- expected = '''<domain type="xen">
- <name>test-inst-test</name>
- <uuid>7db78950-c467-4863-94d1-af59806384ea</uuid>
- <memory unit="MiB">500</memory>
- <currentMemory unit="MiB">400</currentMemory>
- <vcpu placement="static">2</vcpu>
- <os>
- <type arch="x86_64" machine="xenpv">linux</type>
- <kernel>/tmp/kernel/vmlinuz</kernel>
- <initrd>/tmp/kernel/initramfs</initrd>
- <cmdline>root=/dev/mapper/dmroot ro nomodeset console=hvc0 rd_NO_PLYMOUTH rd.plymouth.enable=0 plymouth.enable=0 nopat</cmdline>
- </os>
- <features>
- </features>
- <clock offset='utc' adjustment='reset'>
- <timer name="tsc" mode="native"/>
- </clock>
- <on_poweroff>destroy</on_poweroff>
- <on_reboot>destroy</on_reboot>
- <on_crash>destroy</on_crash>
- <devices>
- <disk type="block" device="disk">
- <driver name="phy" />
- <source dev="/tmp/kernel/modules.img" />
- <target dev="xvdd" />
- <backenddomain name="dom0" />
- </disk>
- <console type="pty">
- <target type="xen" port="0"/>
- </console>
- </devices>
- </domain>
- '''
- my_uuid = '7db78950-c467-4863-94d1-af59806384ea'
- vm = self.get_vm(uuid=my_uuid)
- vm.netvm = None
- vm.virt_mode = 'pv'
- # tests for storage are later
- vm.volumes['kernel'] = unittest.mock.Mock(**{
- 'kernels_dir': '/tmp/kernel',
- 'block_device.return_value.domain': 'dom0',
- 'block_device.return_value.script': None,
- 'block_device.return_value.path': '/tmp/kernel/modules.img',
- 'block_device.return_value.devtype': 'disk',
- 'block_device.return_value.name': 'kernel',
- })
- libvirt_xml = vm.create_config_file()
- self.assertXMLEqual(lxml.etree.XML(libvirt_xml),
- lxml.etree.XML(expected))
- def test_600_libvirt_xml_hvm(self):
- expected = '''<domain type="xen">
- <name>test-inst-test</name>
- <uuid>7db78950-c467-4863-94d1-af59806384ea</uuid>
- <memory unit="MiB">500</memory>
- <currentMemory unit="MiB">400</currentMemory>
- <vcpu placement="static">2</vcpu>
- <cpu mode='host-passthrough'>
- <!-- disable nested HVM -->
- <feature name='vmx' policy='disable'/>
- <feature name='svm' policy='disable'/>
- <!-- disable SMAP inside VM, because of Linux bug -->
- <feature name='smap' policy='disable'/>
- </cpu>
- <os>
- <type arch="x86_64" machine="xenfv">hvm</type>
- <!--
- For the libxl backend libvirt switches between OVMF (UEFI)
- and SeaBIOS based on the loader type. This has nothing to
- do with the hvmloader binary.
- -->
- <loader type="rom">hvmloader</loader>
- <boot dev="cdrom" />
- <boot dev="hd" />
- <!-- server_ip is the address of stubdomain. It hosts it's own DNS server. -->
- <cmdline>root=/dev/mapper/dmroot ro nomodeset console=hvc0 rd_NO_PLYMOUTH rd.plymouth.enable=0 plymouth.enable=0 nopat</cmdline>
- </os>
- <features>
- <pae/>
- <acpi/>
- <apic/>
- <viridian/>
- </features>
- <clock offset="variable" adjustment="0" basis="localtime" />
- <on_poweroff>destroy</on_poweroff>
- <on_reboot>destroy</on_reboot>
- <on_crash>destroy</on_crash>
- <devices>
- <emulator type="stubdom-linux" />
- <input type="tablet" bus="usb"/>
- <video>
- <model type="vga"/>
- </video>
- <graphics type="qubes"/>
- </devices>
- </domain>
- '''
- my_uuid = '7db78950-c467-4863-94d1-af59806384ea'
- vm = self.get_vm(uuid=my_uuid)
- vm.netvm = None
- vm.virt_mode = 'hvm'
- libvirt_xml = vm.create_config_file()
- self.assertXMLEqual(lxml.etree.XML(libvirt_xml),
- lxml.etree.XML(expected))
- def test_600_libvirt_xml_pvh(self):
- expected = '''<domain type="xen">
- <name>test-inst-test</name>
- <uuid>7db78950-c467-4863-94d1-af59806384ea</uuid>
- <memory unit="MiB">500</memory>
- <currentMemory unit="MiB">400</currentMemory>
- <vcpu placement="static">2</vcpu>
- <cpu mode='host-passthrough'>
- <!-- disable nested HVM -->
- <feature name='vmx' policy='disable'/>
- <feature name='svm' policy='disable'/>
- <!-- disable SMAP inside VM, because of Linux bug -->
- <feature name='smap' policy='disable'/>
- </cpu>
- <os>
- <type arch="x86_64" machine="xenfv">hvm</type>
- <kernel>/tmp/kernel/vmlinuz</kernel>
- <initrd>/tmp/kernel/initramfs</initrd>
- <cmdline>root=/dev/mapper/dmroot ro nomodeset console=hvc0 rd_NO_PLYMOUTH rd.plymouth.enable=0 plymouth.enable=0 nopat</cmdline>
- </os>
- <features>
- <pae/>
- <acpi/>
- <apic/>
- <viridian/>
- </features>
- <clock offset='utc' adjustment='reset'>
- <timer name="tsc" mode="native"/>
- </clock>
- <on_poweroff>destroy</on_poweroff>
- <on_reboot>destroy</on_reboot>
- <on_crash>destroy</on_crash>
- <devices>
- <disk type="block" device="disk">
- <driver name="phy" />
- <source dev="/tmp/kernel/modules.img" />
- <target dev="xvdd" />
- <backenddomain name="dom0" />
- </disk>
- <emulator type="none" />
- <console type="pty">
- <target type="xen" port="0"/>
- </console>
- </devices>
- </domain>
- '''
- my_uuid = '7db78950-c467-4863-94d1-af59806384ea'
- vm = self.get_vm(uuid=my_uuid)
- vm.netvm = None
- vm.virt_mode = 'pvh'
- # tests for storage are later
- vm.volumes['kernel'] = unittest.mock.Mock(**{
- 'kernels_dir': '/tmp/kernel',
- 'block_device.return_value.domain': 'dom0',
- 'block_device.return_value.script': None,
- 'block_device.return_value.path': '/tmp/kernel/modules.img',
- 'block_device.return_value.devtype': 'disk',
- 'block_device.return_value.name': 'kernel',
- })
- libvirt_xml = vm.create_config_file()
- self.assertXMLEqual(lxml.etree.XML(libvirt_xml),
- lxml.etree.XML(expected))
- @unittest.mock.patch('qubes.utils.get_timezone')
- @unittest.mock.patch('qubes.utils.urandom')
- @unittest.mock.patch('qubes.vm.qubesvm.QubesVM.untrusted_qdb')
- def test_620_qdb_standalone(self, mock_qubesdb, mock_urandom,
- mock_timezone):
- mock_urandom.return_value = b'A' * 64
- mock_timezone.return_value = 'UTC'
- vm = self.get_vm(cls=qubes.vm.standalonevm.StandaloneVM)
- vm.netvm = None
- vm.events_enabled = True
- test_qubesdb = TestQubesDB()
- mock_qubesdb.write.side_effect = test_qubesdb.write
- mock_qubesdb.rm.side_effect = test_qubesdb.rm
- vm.create_qdb_entries()
- self.maxDiff = None
- iptables_header = (
- '# Generated by Qubes Core on {}\n'
- '*filter\n'
- ':INPUT DROP [0:0]\n'
- ':FORWARD DROP [0:0]\n'
- ':OUTPUT ACCEPT [0:0]\n'
- '-A INPUT -i vif+ -p udp -m udp --dport 68 -j DROP\n'
- '-A INPUT -m conntrack --ctstate '
- 'RELATED,ESTABLISHED -j ACCEPT\n'
- '-A INPUT -p icmp -j ACCEPT\n'
- '-A INPUT -i lo -j ACCEPT\n'
- '-A INPUT -j REJECT --reject-with '
- 'icmp-host-prohibited\n'
- '-A FORWARD -m conntrack --ctstate '
- 'RELATED,ESTABLISHED -j ACCEPT\n'
- '-A FORWARD -i vif+ -o vif+ -j DROP\n'
- 'COMMIT\n'.format(datetime.datetime.now().ctime()))
- self.assertEqual(test_qubesdb.data, {
- '/name': 'test-inst-test',
- '/type': 'StandaloneVM',
- '/qubes-vm-type': 'AppVM',
- '/qubes-debug-mode': '0',
- '/qubes-base-template': '',
- '/qubes-timezone': 'UTC',
- '/qubes-random-seed': base64.b64encode(b'A' * 64),
- '/qubes-vm-persistence': 'full',
- '/qubes-vm-updateable': 'True',
- '/qubes-block-devices': '',
- '/qubes-usb-devices': '',
- '/qubes-iptables': 'reload',
- '/qubes-iptables-error': '',
- '/qubes-iptables-header': iptables_header,
- '/qubes-service/qubes-update-check': '0',
- })
- @unittest.mock.patch('qubes.utils.get_timezone')
- @unittest.mock.patch('qubes.utils.urandom')
- @unittest.mock.patch('qubes.vm.qubesvm.QubesVM.untrusted_qdb')
- def test_621_qdb_appvm_with_network(self, mock_qubesdb, mock_urandom,
- mock_timezone):
- mock_urandom.return_value = b'A' * 64
- mock_timezone.return_value = 'UTC'
- template = self.get_vm(cls=qubes.vm.templatevm.TemplateVM, name='template')
- template.netvm = None
- netvm = self.get_vm(cls=qubes.vm.standalonevm.StandaloneVM,
- name='netvm', qid=2, provides_network=True)
- vm = self.get_vm(cls=qubes.vm.appvm.AppVM, template=template,
- name='appvm', qid=3)
- vm.netvm = netvm
- test_qubesdb = TestQubesDB()
- mock_qubesdb.write.side_effect = test_qubesdb.write
- mock_qubesdb.rm.side_effect = test_qubesdb.rm
- self.maxDiff = None
- iptables_header = (
- '# Generated by Qubes Core on {}\n'
- '*filter\n'
- ':INPUT DROP [0:0]\n'
- ':FORWARD DROP [0:0]\n'
- ':OUTPUT ACCEPT [0:0]\n'
- '-A INPUT -i vif+ -p udp -m udp --dport 68 -j DROP\n'
- '-A INPUT -m conntrack --ctstate '
- 'RELATED,ESTABLISHED -j ACCEPT\n'
- '-A INPUT -p icmp -j ACCEPT\n'
- '-A INPUT -i lo -j ACCEPT\n'
- '-A INPUT -j REJECT --reject-with '
- 'icmp-host-prohibited\n'
- '-A FORWARD -m conntrack --ctstate '
- 'RELATED,ESTABLISHED -j ACCEPT\n'
- '-A FORWARD -i vif+ -o vif+ -j DROP\n'
- 'COMMIT\n'.format(datetime.datetime.now().ctime()))
- expected = {
- '/name': 'test-inst-appvm',
- '/type': 'AppVM',
- '/qubes-vm-type': 'AppVM',
- '/qubes-debug-mode': '0',
- '/qubes-base-template': 'test-inst-template',
- '/qubes-timezone': 'UTC',
- '/qubes-random-seed': base64.b64encode(b'A' * 64),
- '/qubes-vm-persistence': 'rw-only',
- '/qubes-vm-updateable': 'False',
- '/qubes-block-devices': '',
- '/qubes-usb-devices': '',
- '/qubes-iptables': 'reload',
- '/qubes-iptables-error': '',
- '/qubes-iptables-header': iptables_header,
- '/qubes-service/qubes-update-check': '0',
- '/qubes-ip': '10.137.0.3',
- '/qubes-netmask': '255.255.255.255',
- '/qubes-gateway': '10.137.0.2',
- '/qubes-primary-dns': '10.139.1.1',
- '/qubes-secondary-dns': '10.139.1.2',
- }
- vm.create_qdb_entries()
- self.assertEqual(test_qubesdb.data, expected)
|