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

Follow change of qubesdb path return type

Browse Source 
It's str/unicode, not bytes now.

QubesOS/qubes-issues#2937
This commit is contained in:
Marek Marczykowski-Górecki 2017-07-24 13:02:41 +02:00
parent 1759bca00f
commit 0fb7c1fbed
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
4 changed files with 80 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
qubes/api/misc.py
View File

@ -51,8 +51,7 @@ class QubesMiscAPI(qubes.api.AbstractQubesAPI):
prefix = '/features-request/' prefix = '/features-request/'
keys = [key.decode('ascii', errors='strict') keys = self.src.untrusted_qdb.list(prefix)
for key in self.src.untrusted_qdb.list(prefix)]
untrusted_features = {key[len(prefix):]: untrusted_features = {key[len(prefix):]:
self.src.untrusted_qdb.read(key).decode('ascii', errors='strict') self.src.untrusted_qdb.read(key).decode('ascii', errors='strict')
for key in keys} for key in keys}

7
qubes/ext/block.py
View File

@ -124,12 +124,11 @@ class BlockDeviceExtension(qubes.ext.Extension):
def on_device_list_block(self, vm, event): def on_device_list_block(self, vm, event):
# pylint: disable=unused-argument,no-self-use # pylint: disable=unused-argument,no-self-use
safe_set = {ord(c) for c in safe_set = string.ascii_letters + string.digits
string.ascii_letters + string.digits}
if not vm.is_running(): if not vm.is_running():
return return
untrusted_qubes_devices = vm.untrusted_qdb.list('/qubes-block-devices/') untrusted_qubes_devices = vm.untrusted_qdb.list('/qubes-block-devices/')
untrusted_idents = set(untrusted_path.split(b'/', 3)[2] untrusted_idents = set(untrusted_path.split('/', 3)[2]
for untrusted_path in untrusted_qubes_devices) for untrusted_path in untrusted_qubes_devices)
for untrusted_ident in untrusted_idents: for untrusted_ident in untrusted_idents:
if not all(c in safe_set for c in untrusted_ident): if not all(c in safe_set for c in untrusted_ident):
@ -138,7 +137,7 @@ class BlockDeviceExtension(qubes.ext.Extension):
vm.log.warning(msg % vm.name) vm.log.warning(msg % vm.name)
continue continue
ident = untrusted_ident.decode('ascii', errors='strict') ident = untrusted_ident
device_info = self.device_get(vm, ident) device_info = self.device_get(vm, ident)
if device_info: if device_info:

2
qubes/tests/api_misc.py
View File

@ -40,7 +40,7 @@ class TC_00_API_Misc(qubes.tests.QubesTestCase):
'untrusted_qdb.read.side_effect': ( 'untrusted_qdb.read.side_effect': (
lambda path: entries.get(path, None)), lambda path: entries.get(path, None)),
'untrusted_qdb.list.side_effect': ( 'untrusted_qdb.list.side_effect': (
lambda path: sorted(map(str.encode, entries.keys()))), lambda path: sorted(entries.keys())),
}) })
def call_mgmt_func(self, method, arg=b'', payload=b''): def call_mgmt_func(self, method, arg=b'', payload=b''):

154
qubes/tests/devices_block.py
View File

@ -94,13 +94,9 @@ class TestQubesDB(object):
self._data = data self._data = data
def read(self, key): def read(self, key):
if isinstance(key, str):
key = key.encode()
return self._data.get(key, None) return self._data.get(key, None)
def list(self, prefix): def list(self, prefix):
if isinstance(prefix, str):
prefix = prefix.encode()
return [key for key in self._data if key.startswith(prefix)] return [key for key in self._data if key.startswith(prefix)]
@ -143,10 +139,10 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_000_device_get(self): def test_000_device_get(self):
vm = TestVM({ vm = TestVM({
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'w', '/qubes-block-devices/sda/mode': b'w',
}) })
device_info = self.ext.device_get(vm, 'sda') device_info = self.ext.device_get(vm, 'sda')
self.assertIsInstance(device_info, qubes.ext.block.BlockDevice) self.assertIsInstance(device_info, qubes.ext.block.BlockDevice)
@ -161,10 +157,10 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_001_device_get_other_node(self): def test_001_device_get_other_node(self):
vm = TestVM({ vm = TestVM({
b'/qubes-block-devices/mapper_dmroot': b'', '/qubes-block-devices/mapper_dmroot': b'',
b'/qubes-block-devices/mapper_dmroot/desc': b'Test device', '/qubes-block-devices/mapper_dmroot/desc': b'Test device',
b'/qubes-block-devices/mapper_dmroot/size': b'1024000', '/qubes-block-devices/mapper_dmroot/size': b'1024000',
b'/qubes-block-devices/mapper_dmroot/mode': b'w', '/qubes-block-devices/mapper_dmroot/mode': b'w',
}) })
device_info = self.ext.device_get(vm, 'mapper_dmroot') device_info = self.ext.device_get(vm, 'mapper_dmroot')
self.assertIsInstance(device_info, qubes.ext.block.BlockDevice) self.assertIsInstance(device_info, qubes.ext.block.BlockDevice)
@ -179,20 +175,20 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_002_device_get_invalid_desc(self): def test_002_device_get_invalid_desc(self):
vm = TestVM({ vm = TestVM({
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device<>za\xc4\x87abc', '/qubes-block-devices/sda/desc': b'Test device<>za\xc4\x87abc',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'w', '/qubes-block-devices/sda/mode': b'w',
}) })
device_info = self.ext.device_get(vm, 'sda') device_info = self.ext.device_get(vm, 'sda')
self.assertEqual(device_info.description, 'Test device__za__abc') self.assertEqual(device_info.description, 'Test device__za__abc')
def test_003_device_get_invalid_size(self): def test_003_device_get_invalid_size(self):
vm = TestVM({ vm = TestVM({
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000abc', '/qubes-block-devices/sda/size': b'1024000abc',
b'/qubes-block-devices/sda/mode': b'w', '/qubes-block-devices/sda/mode': b'w',
}) })
device_info = self.ext.device_get(vm, 'sda') device_info = self.ext.device_get(vm, 'sda')
self.assertEqual(device_info.size, 0) self.assertEqual(device_info.size, 0)
@ -200,10 +196,10 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_004_device_get_invalid_mode(self): def test_004_device_get_invalid_mode(self):
vm = TestVM({ vm = TestVM({
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'abc', '/qubes-block-devices/sda/mode': b'abc',
}) })
device_info = self.ext.device_get(vm, 'sda') device_info = self.ext.device_get(vm, 'sda')
self.assertEqual(device_info.mode, 'w') self.assertEqual(device_info.mode, 'w')
@ -211,24 +207,24 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_005_device_get_none(self): def test_005_device_get_none(self):
vm = TestVM({ vm = TestVM({
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'w', '/qubes-block-devices/sda/mode': b'w',
}) })
device_info = self.ext.device_get(vm, 'sdb') device_info = self.ext.device_get(vm, 'sdb')
self.assertIsNone(device_info) self.assertIsNone(device_info)
def test_010_devices_list(self): def test_010_devices_list(self):
vm = TestVM({ vm = TestVM({
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'w', '/qubes-block-devices/sda/mode': b'w',
b'/qubes-block-devices/sdb': b'', '/qubes-block-devices/sdb': b'',
b'/qubes-block-devices/sdb/desc': b'Test device2', '/qubes-block-devices/sdb/desc': b'Test device2',
b'/qubes-block-devices/sdb/size': b'2048000', '/qubes-block-devices/sdb/size': b'2048000',
b'/qubes-block-devices/sdb/mode': b'r', '/qubes-block-devices/sdb/mode': b'r',
}) })
devices = sorted(list(self.ext.on_device_list_block(vm, ''))) devices = sorted(list(self.ext.on_device_list_block(vm, '')))
self.assertEqual(len(devices), 2) self.assertEqual(len(devices), 2)
@ -250,9 +246,9 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_012_devices_list_invalid_ident(self): def test_012_devices_list_invalid_ident(self):
vm = TestVM({ vm = TestVM({
b'/qubes-block-devices/invalid ident': b'', '/qubes-block-devices/invalid ident': b'',
b'/qubes-block-devices/invalid+ident': b'', '/qubes-block-devices/invalid+ident': b'',
b'/qubes-block-devices/invalid#': b'', '/qubes-block-devices/invalid#': b'',
}) })
devices = sorted(list(self.ext.on_device_list_block(vm, ''))) devices = sorted(list(self.ext.on_device_list_block(vm, '')))
self.assertEqual(len(devices), 0) self.assertEqual(len(devices), 0)
@ -334,10 +330,10 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_040_attach(self): def test_040_attach(self):
back_vm = TestVM(name='sys-usb', qdb={ back_vm = TestVM(name='sys-usb', qdb={
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'w', '/qubes-block-devices/sda/mode': b'w',
}) })
vm = TestVM({}, domain_xml=domain_xml_template.format('')) vm = TestVM({}, domain_xml=domain_xml_template.format(''))
dev = qubes.ext.block.BlockDevice(back_vm, 'sda') dev = qubes.ext.block.BlockDevice(back_vm, 'sda')
@ -353,10 +349,10 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_041_attach_frontend(self): def test_041_attach_frontend(self):
back_vm = TestVM(name='sys-usb', qdb={ back_vm = TestVM(name='sys-usb', qdb={
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'w', '/qubes-block-devices/sda/mode': b'w',
}) })
vm = TestVM({}, domain_xml=domain_xml_template.format('')) vm = TestVM({}, domain_xml=domain_xml_template.format(''))
dev = qubes.ext.block.BlockDevice(back_vm, 'sda') dev = qubes.ext.block.BlockDevice(back_vm, 'sda')
@ -373,10 +369,10 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_042_attach_read_only(self): def test_042_attach_read_only(self):
back_vm = TestVM(name='sys-usb', qdb={ back_vm = TestVM(name='sys-usb', qdb={
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'w', '/qubes-block-devices/sda/mode': b'w',
}) })
vm = TestVM({}, domain_xml=domain_xml_template.format('')) vm = TestVM({}, domain_xml=domain_xml_template.format(''))
dev = qubes.ext.block.BlockDevice(back_vm, 'sda') dev = qubes.ext.block.BlockDevice(back_vm, 'sda')
@ -394,10 +390,10 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_043_attach_invalid_option(self): def test_043_attach_invalid_option(self):
back_vm = TestVM(name='sys-usb', qdb={ back_vm = TestVM(name='sys-usb', qdb={
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'w', '/qubes-block-devices/sda/mode': b'w',
}) })
vm = TestVM({}, domain_xml=domain_xml_template.format('')) vm = TestVM({}, domain_xml=domain_xml_template.format(''))
dev = qubes.ext.block.BlockDevice(back_vm, 'sda') dev = qubes.ext.block.BlockDevice(back_vm, 'sda')
@ -408,10 +404,10 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_044_attach_invalid_option2(self): def test_044_attach_invalid_option2(self):
back_vm = TestVM(name='sys-usb', qdb={ back_vm = TestVM(name='sys-usb', qdb={
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'w', '/qubes-block-devices/sda/mode': b'w',
}) })
vm = TestVM({}, domain_xml=domain_xml_template.format('')) vm = TestVM({}, domain_xml=domain_xml_template.format(''))
dev = qubes.ext.block.BlockDevice(back_vm, 'sda') dev = qubes.ext.block.BlockDevice(back_vm, 'sda')
@ -422,10 +418,10 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_045_attach_backend_not_running(self): def test_045_attach_backend_not_running(self):
back_vm = TestVM(name='sys-usb', running=False, qdb={ back_vm = TestVM(name='sys-usb', running=False, qdb={
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'w', '/qubes-block-devices/sda/mode': b'w',
}) })
vm = TestVM({}, domain_xml=domain_xml_template.format('')) vm = TestVM({}, domain_xml=domain_xml_template.format(''))
dev = qubes.ext.block.BlockDevice(back_vm, 'sda') dev = qubes.ext.block.BlockDevice(back_vm, 'sda')
@ -435,10 +431,10 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_046_attach_ro_dev_rw(self): def test_046_attach_ro_dev_rw(self):
back_vm = TestVM(name='sys-usb', qdb={ back_vm = TestVM(name='sys-usb', qdb={
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'r', '/qubes-block-devices/sda/mode': b'r',
}) })
vm = TestVM({}, domain_xml=domain_xml_template.format('')) vm = TestVM({}, domain_xml=domain_xml_template.format(''))
dev = qubes.ext.block.BlockDevice(back_vm, 'sda') dev = qubes.ext.block.BlockDevice(back_vm, 'sda')
@ -449,10 +445,10 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_047_attach_read_only_auto(self): def test_047_attach_read_only_auto(self):
back_vm = TestVM(name='sys-usb', qdb={ back_vm = TestVM(name='sys-usb', qdb={
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'r', '/qubes-block-devices/sda/mode': b'r',
}) })
vm = TestVM({}, domain_xml=domain_xml_template.format('')) vm = TestVM({}, domain_xml=domain_xml_template.format(''))
dev = qubes.ext.block.BlockDevice(back_vm, 'sda') dev = qubes.ext.block.BlockDevice(back_vm, 'sda')
@ -469,10 +465,10 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_050_detach(self): def test_050_detach(self):
back_vm = TestVM(name='sys-usb', qdb={ back_vm = TestVM(name='sys-usb', qdb={
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'r', '/qubes-block-devices/sda/mode': b'r',
}) })
device_xml = ( device_xml = (
'<disk type="block" device="disk">\n' '<disk type="block" device="disk">\n'
@ -491,10 +487,10 @@ class TC_00_Block(qubes.tests.QubesTestCase):
def test_051_detach_not_attached(self): def test_051_detach_not_attached(self):
back_vm = TestVM(name='sys-usb', qdb={ back_vm = TestVM(name='sys-usb', qdb={
b'/qubes-block-devices/sda': b'', '/qubes-block-devices/sda': b'',
b'/qubes-block-devices/sda/desc': b'Test device', '/qubes-block-devices/sda/desc': b'Test device',
b'/qubes-block-devices/sda/size': b'1024000', '/qubes-block-devices/sda/size': b'1024000',
b'/qubes-block-devices/sda/mode': b'r', '/qubes-block-devices/sda/mode': b'r',
}) })
device_xml = ( device_xml = (
'<disk type="block" device="disk">\n' '<disk type="block" device="disk">\n'