Qubes/core-admin
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

api/admin: implement admin.vm.device....Set.persistent

Browse Source 
This will allow converting persistent device (used to boot VM for
example) to non-persistent.

QubesOS/qubes-issues#3055
This commit is contained in:
Marek Marczykowski-Górecki 2017-09-02 17:59:29 +02:00
parent 9d062c4c66
commit cea70748a6
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
3 changed files with 130 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Makefile
View File

@ -55,14 +55,17 @@ ADMIN_API_METHODS_SIMPLE = \
admin.vm.device.pci.Available \
admin.vm.device.pci.Detach \
admin.vm.device.pci.List \
admin.vm.device.pci.Set.persistent \
admin.vm.device.block.Attach \
admin.vm.device.block.Available \
admin.vm.device.block.Detach \
admin.vm.device.block.List \
admin.vm.device.block.Set.persistent \
admin.vm.device.mic.Attach \
admin.vm.device.mic.Available \
admin.vm.device.mic.Detach \
admin.vm.device.mic.List \
admin.vm.device.mic.Set.persistent \
admin.vm.feature.CheckWithTemplate \
admin.vm.feature.Get \
admin.vm.feature.List \
@ -188,6 +191,7 @@ endif
admin.vm.device.testclass.Attach \
admin.vm.device.testclass.Detach \
admin.vm.device.testclass.List \
admin.vm.device.testclass.Set.persistent \
admin.vm.device.testclass.Available
# sanity check
for method in $(DESTDIR)/etc/qubes-rpc/policy/admin.*; do \

30
qubes/api/admin.py
View File

@ -1093,6 +1093,36 @@ class QubesAdminAPI(qubes.api.AbstractQubesAPI):
yield from self.dest.devices[devclass].detach(assignment)
self.app.save()
# Attach/Detach action can both modify persistent state (with
# persistent=True) and volatile state of running VM (with persistent=False).
# For this reason, write=True + execute=True
@qubes.api.method('admin.vm.device.{endpoint}.Set.persistent',
endpoints=(ep.name
for ep in pkg_resources.iter_entry_points('qubes.devices')),
scope='local', write=True, execute=True)
@asyncio.coroutine
def vm_device_set_persistent(self, endpoint, untrusted_payload):
devclass = endpoint
assert untrusted_payload in (b'True', b'False')
persistent = untrusted_payload == b'True'
del untrusted_payload
# qrexec already verified that no strange characters are in self.arg
backend_domain, ident = self.arg.split('+', 1)
# device must be already attached
matching_devices = [dev for dev
in self.dest.devices[devclass].attached()
if dev.backend_domain.name == backend_domain and dev.ident == ident]
assert len(matching_devices) == 1
dev = matching_devices[0]
self.fire_event_for_permission(device=dev,
persistent=persistent)
self.dest.devices[devclass].update_persistent(dev, persistent)
self.app.save()
@qubes.api.method('admin.vm.firewall.Get', no_payload=True,
scope='local', read=True)
@asyncio.coroutine

96
qubes/tests/api_admin.py
View File

@ -2131,6 +2131,102 @@ class TC_00_VMs(AdminAPITestCase):
b'test-vm1')
self.assertFalse(self.app.save.called)
def test_650_vm_device_set_persistent_true(self):
self.vm.add_handler('device-list:testclass',
self.device_list_testclass)
self.vm.add_handler('device-list-attached:testclass',
self.device_list_attached_testclass)
with unittest.mock.patch.object(qubes.vm.qubesvm.QubesVM,
'is_halted', lambda _: False):
value = self.call_mgmt_func(
b'admin.vm.device.testclass.Set.persistent',
b'test-vm1', b'test-vm1+1234', b'True')
self.assertIsNone(value)
dev = qubes.devices.DeviceInfo(self.vm, '1234')
self.assertIn(dev, self.vm.devices['testclass'].persistent())
self.app.save.assert_called_once_with()
def test_651_vm_device_set_persistent_false_unchanged(self):
self.vm.add_handler('device-list:testclass',
self.device_list_testclass)
self.vm.add_handler('device-list-attached:testclass',
self.device_list_attached_testclass)
with unittest.mock.patch.object(qubes.vm.qubesvm.QubesVM,
'is_halted', lambda _: False):
value = self.call_mgmt_func(
b'admin.vm.device.testclass.Set.persistent',
b'test-vm1', b'test-vm1+1234', b'False')
self.assertIsNone(value)
dev = qubes.devices.DeviceInfo(self.vm, '1234')
self.assertNotIn(dev, self.vm.devices['testclass'].persistent())
self.app.save.assert_called_once_with()
def test_652_vm_device_set_persistent_false(self):
self.vm.add_handler('device-list:testclass',
self.device_list_testclass)
assignment = qubes.devices.DeviceAssignment(self.vm, '1234', {},
True)
self.loop.run_until_complete(
self.vm.devices['testclass'].attach(assignment))
self.vm.add_handler('device-list-attached:testclass',
self.device_list_attached_testclass)
dev = qubes.devices.DeviceInfo(self.vm, '1234')
self.assertIn(dev, self.vm.devices['testclass'].persistent())
with unittest.mock.patch.object(qubes.vm.qubesvm.QubesVM,
'is_halted', lambda _: False):
value = self.call_mgmt_func(
b'admin.vm.device.testclass.Set.persistent',
b'test-vm1', b'test-vm1+1234', b'False')
self.assertIsNone(value)
self.assertNotIn(dev, self.vm.devices['testclass'].persistent())
self.assertIn(dev, self.vm.devices['testclass'].attached())
self.app.save.assert_called_once_with()
def test_653_vm_device_set_persistent_true_unchanged(self):
self.vm.add_handler('device-list:testclass',
self.device_list_testclass)
assignment = qubes.devices.DeviceAssignment(self.vm, '1234', {},
True)
self.loop.run_until_complete(
self.vm.devices['testclass'].attach(assignment))
self.vm.add_handler('device-list-attached:testclass',
self.device_list_attached_testclass)
with unittest.mock.patch.object(qubes.vm.qubesvm.QubesVM,
'is_halted', lambda _: False):
value = self.call_mgmt_func(
b'admin.vm.device.testclass.Set.persistent',
b'test-vm1', b'test-vm1+1234', b'True')
self.assertIsNone(value)
dev = qubes.devices.DeviceInfo(self.vm, '1234')
self.assertIn(dev, self.vm.devices['testclass'].persistent())
self.assertIn(dev, self.vm.devices['testclass'].attached())
self.app.save.assert_called_once_with()
def test_654_vm_device_set_persistent_not_attached(self):
self.vm.add_handler('device-list:testclass',
self.device_list_testclass)
with unittest.mock.patch.object(qubes.vm.qubesvm.QubesVM,
'is_halted', lambda _: False):
with self.assertRaises(AssertionError):
self.call_mgmt_func(
b'admin.vm.device.testclass.Set.persistent',
b'test-vm1', b'test-vm1+1234', b'True')
dev = qubes.devices.DeviceInfo(self.vm, '1234')
self.assertNotIn(dev, self.vm.devices['testclass'].persistent())
self.assertFalse(self.app.save.called)
def test_655_vm_device_set_persistent_invalid_value(self):
self.vm.add_handler('device-list:testclass',
self.device_list_testclass)
with unittest.mock.patch.object(qubes.vm.qubesvm.QubesVM,
'is_halted', lambda _: False):
with self.assertRaises(AssertionError):
self.call_mgmt_func(
b'admin.vm.device.testclass.Set.persistent',
b'test-vm1', b'test-vm1+1234', b'maybe')
dev = qubes.devices.DeviceInfo(self.vm, '1234')
self.assertNotIn(dev, self.vm.devices['testclass'].persistent())
self.assertFalse(self.app.save.called)
def test_990_vm_unexpected_payload(self):
methods_with_no_payload = [