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

Merge branch 'dispvm-policy'

Browse Source
This commit is contained in:
Marek Marczykowski-Górecki 2017-09-14 00:15:29 +02:00
commit 55208a0c7c
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
13 changed files with 280 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
doc/qubes-policy.rst
View File

@ -29,6 +29,7 @@ Each line consist of three values separated by white characters (space(s), tab(s
AppVM, TemplateVM, StandaloneVM, DispVM
- `$default` - used when caller did not specified any VM
- `$dispvm:vm-name` - _new_ Disposable VM created from AppVM `vm-name`
- `$dispvm:$tag:some-tag` - _new_ Disposable VM created from AppVM tagged with `some-tag`
- `$dispvm` - _new_ Disposable VM created from AppVM pointed by caller
property `default_dispvm`, which defaults to global property `default_dispvm`
- `$adminvm` - Admin VM aka dom0

2
qubes/__init__.py
View File

@ -640,7 +640,7 @@ class PropertyHolder(qubes.events.Emitter):
'''Clone properties from other object.
:param PropertyHolder src: source object
:param list proplist: list of properties \
:param iterable proplist: list of properties \
(:py:obj:`None` or omit for all properties except those with \
:py:attr:`property.clone` set to :py:obj:`False`)
'''

3
qubes/api/internal.py
View File

@ -46,7 +46,8 @@ class QubesInternalAPI(qubes.api.AbstractQubesAPI):
domain.name: {
'tags': list(domain.tags),
'type': domain.__class__.__name__,
'dispvm_allowed': getattr(domain, 'dispvm_allowed', False),
'template_for_dispvms':
getattr(domain, 'template_for_dispvms', False),
'default_dispvm': (str(domain.default_dispvm) if
getattr(domain, 'default_dispvm', None) else None),
'icon': str(domain.label.icon),

4
qubes/tests/api_admin.py
View File

@ -2102,7 +2102,7 @@ class TC_00_VMs(AdminAPITestCase):
@unittest.mock.patch('qubes.storage.Storage.create')
def test_640_vm_create_disposable(self, mock_storage):
mock_storage.side_effect = self.dummy_coro
self.vm.dispvm_allowed = True
self.vm.template_for_dispvms = True
retval = self.call_mgmt_func(b'admin.vm.CreateDisposable',
b'test-vm1')
self.assertTrue(retval.startswith('disp'))
@ -2115,7 +2115,7 @@ class TC_00_VMs(AdminAPITestCase):
@unittest.mock.patch('qubes.storage.Storage.create')
def test_641_vm_create_disposable_default(self, mock_storage):
mock_storage.side_effect = self.dummy_coro
self.vm.dispvm_allowed = True
self.vm.template_for_dispvms = True
self.app.default_dispvm = self.vm
retval = self.call_mgmt_func(b'admin.vm.CreateDisposable',
b'dom0')

25
qubes/tests/vm/appvm.py
View File

@ -20,6 +20,8 @@
from unittest import mock
import lxml.etree
import qubes.storage
import qubes.tests
import qubes.tests.vm.qubesvm
@ -118,3 +120,26 @@ class TC_90_AppVM(qubes.tests.vm.qubesvm.QubesVMTestsMixin,
mock_power.return_value = 'Running'
with self.assertRaises(qubes.exc.QubesVMNotHaltedError):
vm.template = self.template
def test_500_property_migrate_template_for_dispvms(self):
xml_template = '''
<domain class="AppVM" id="domain-1">
<properties>
<property name="qid">1</property>
<property name="name">testvm</property>
<property name="label" ref="label-1" />
<property name="dispvm_allowed">{value}</property>
</properties>
</domain>
'''
xml = lxml.etree.XML(xml_template.format(value='True'))
vm = qubes.vm.appvm.AppVM(self.app, xml)
self.assertEqual(vm.template_for_dispvms, True)
with self.assertRaises(AttributeError):
vm.dispvm_allowed
xml = lxml.etree.XML(xml_template.format(value='False'))
vm = qubes.vm.appvm.AppVM(self.app, xml)
self.assertEqual(vm.template_for_dispvms, False)
with self.assertRaises(AttributeError):
vm.dispvm_allowed

52
qubes/tests/vm/dispvm.py
View File

@ -67,7 +67,7 @@ class TC_00_DispVM(qubes.tests.QubesTestCase):
@mock.patch('qubes.storage.Storage')
def test_000_from_appvm(self, mock_storage, mock_makedirs, mock_symlink):
mock_storage.return_value.create.side_effect = self.mock_coro
self.appvm.dispvm_allowed = True
self.appvm.template_for_dispvms = True
orig_getitem = self.app.domains.__getitem__
with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
as mock_domains:
@ -78,7 +78,7 @@ class TC_00_DispVM(qubes.tests.QubesTestCase):
dispvm = self.loop.run_until_complete(
qubes.vm.dispvm.DispVM.from_appvm(self.appvm))
mock_domains.get_new_unused_dispid.assert_called_once_with()
self.assertTrue(dispvm.name.startswith('disp'))
self.assertEqual(dispvm.name, 'disp42')
self.assertEqual(dispvm.template, self.appvm)
self.assertEqual(dispvm.label, self.appvm.label)
self.assertEqual(dispvm.label, self.appvm.label)
@ -93,3 +93,51 @@ class TC_00_DispVM(qubes.tests.QubesTestCase):
with self.assertRaises(qubes.exc.QubesException):
dispvm = self.loop.run_until_complete(
qubes.vm.dispvm.DispVM.from_appvm(self.appvm))
def test_010_create_direct(self):
self.appvm.template_for_dispvms = True
orig_getitem = self.app.domains.__getitem__
with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
as mock_domains:
mock_domains.configure_mock(**{
'get_new_unused_dispid': mock.Mock(return_value=42),
'__getitem__.side_effect': orig_getitem
})
dispvm = self.app.add_new_vm(qubes.vm.dispvm.DispVM,
name='test-dispvm', template=self.appvm)
mock_domains.get_new_unused_dispid.assert_called_once_with()
self.assertEqual(dispvm.name, 'test-dispvm')
self.assertEqual(dispvm.template, self.appvm)
self.assertEqual(dispvm.label, self.appvm.label)
self.assertEqual(dispvm.label, self.appvm.label)
self.assertEqual(dispvm.auto_cleanup, False)
def test_011_create_direct_generate_name(self):
self.appvm.template_for_dispvms = True
orig_getitem = self.app.domains.__getitem__
with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
as mock_domains:
mock_domains.configure_mock(**{
'get_new_unused_dispid': mock.Mock(return_value=42),
'__getitem__.side_effect': orig_getitem
})
dispvm = self.app.add_new_vm(qubes.vm.dispvm.DispVM,
template=self.appvm)
mock_domains.get_new_unused_dispid.assert_called_once_with()
self.assertEqual(dispvm.name, 'disp42')
self.assertEqual(dispvm.template, self.appvm)
self.assertEqual(dispvm.label, self.appvm.label)
self.assertEqual(dispvm.auto_cleanup, False)
def test_011_create_direct_reject(self):
orig_getitem = self.app.domains.__getitem__
with mock.patch.object(self.app, 'domains', wraps=self.app.domains) \
as mock_domains:
mock_domains.configure_mock(**{
'get_new_unused_dispid': mock.Mock(return_value=42),
'__getitem__.side_effect': orig_getitem
})
with self.assertRaises(qubes.exc.QubesException):
self.app.add_new_vm(qubes.vm.dispvm.DispVM,
name='test-dispvm', template=self.appvm)
self.assertFalse(mock_domains.get_new_unused_dispid.called)

11
qubes/vm/appvm.py
View File

@ -36,7 +36,7 @@ class AppVM(qubes.vm.qubesvm.QubesVM):
vmclass=qubes.vm.templatevm.TemplateVM,
doc='Template, on which this AppVM is based.')
dispvm_allowed = qubes.property('dispvm_allowed',
template_for_dispvms = qubes.property('template_for_dispvms',
type=bool,
default=False,
doc='Should this VM be allowed to start as Disposable VM')
@ -72,6 +72,15 @@ class AppVM(qubes.vm.qubesvm.QubesVM):
}
def __init__(self, app, xml, **kwargs):
# migrate renamed properties
if xml is not None:
node_dispvm_allowed = xml.find(
'./properties/property[@name=\'dispvm_allowed\']')
if node_dispvm_allowed is not None:
kwargs['template_for_dispvms'] = \
qubes.property.bool(None, None, node_dispvm_allowed.text)
node_dispvm_allowed.getparent().remove(node_dispvm_allowed)
self.volume_config = copy.deepcopy(self.default_volume_config)
template = kwargs.get('template', None)

41
qubes/vm/dispvm.py
View File

@ -77,22 +77,18 @@ class DispVM(qubes.vm.qubesvm.QubesVM):
template = kwargs.get('template', None)
if xml is None:
assert template is not None
if not template.template_for_dispvms:
raise qubes.exc.QubesValueError(
'template for DispVM ({}) needs to have '
'template_for_dispvms=True'.format(template.name))
if 'dispid' not in kwargs:
kwargs['dispid'] = app.domains.get_new_unused_dispid()
if 'name' not in kwargs:
kwargs['name'] = 'disp' + str(kwargs['dispid'])
# by default inherit properties from the DispVM template
proplist = [prop.__name__ for prop in template.property_list()
if prop.clone and prop.__name__ not in ['template']]
self_props = [prop.__name__ for prop in self.property_list()]
for prop in proplist:
if prop not in self_props:
continue
if prop not in kwargs and \
not template.property_is_default(prop):
kwargs[prop] = getattr(template, prop)
if template is not None:
# template is only passed if the AppVM is created, in other cases we
# don't need to patch the volume_config because the config is
@ -108,6 +104,13 @@ class DispVM(qubes.vm.qubesvm.QubesVM):
super(DispVM, self).__init__(app, xml, *args, **kwargs)
if xml is None:
# by default inherit properties from the DispVM template
proplist = [prop.__name__ for prop in template.property_list()
if prop.clone and prop.__name__ not in ['template']]
self_props = [prop.__name__ for prop in self.property_list()]
self.clone_properties(template, set(proplist).intersection(
self_props))
self.firewall.clone(template.firewall)
self.features.update(template.features)
self.tags.update(template.tags)
@ -158,10 +161,10 @@ class DispVM(qubes.vm.qubesvm.QubesVM):
This method modifies :file:`qubes.xml` file.
The qube returned is not started.
'''
if not appvm.dispvm_allowed:
if not appvm.template_for_dispvms:
raise qubes.exc.QubesException(
'Refusing to create DispVM out of this AppVM, because '
'dispvm_allowed=False')
'template_for_appvms=False')
app = appvm.app
dispvm = app.add_new_vm(
cls,
@ -189,3 +192,15 @@ class DispVM(qubes.vm.qubesvm.QubesVM):
yield from self.remove_from_disk()
del self.app.domains[self]
self.app.save()
@asyncio.coroutine
def start(self, **kwargs):
# pylint: disable=arguments-differ
# sanity check, if template_for_dispvm got changed in the meantime
if not self.template.template_for_dispvms:
raise qubes.exc.QubesException(
'template for DispVM ({}) needs to have '
'template_for_dispvms=True'.format(self.template.name))
yield from super(DispVM, self).start(**kwargs)

90
qubespolicy/__init__.py
View File

@ -80,35 +80,48 @@ def verify_target_value(system_info, value):
if dispvm_base not in system_info['domains']:
return False
dispvm_base_info = system_info['domains'][dispvm_base]
return bool(dispvm_base_info['dispvm_allowed'])
return bool(dispvm_base_info['template_for_dispvms'])
else:
return value in system_info['domains']
def verify_special_value(value, for_target=True):
def verify_special_value(value, for_target=True, specific_target=False):
'''
Verify if given special VM-specifier ('$...') is valid
:param value: value to verify
:param for_target: should classify target-only values as valid (
'$default', '$dispvm')
:param specific_target: allow only values naming specific target
(for use with target=, default= etc)
:return: True or False
'''
# pylint: disable=too-many-return-statements
if value.startswith('$tag:') and len(value) > len('$tag:'):
# values used only for matching VMs, not naming specific one (for actual
# call target)
if not specific_target:
if value.startswith('$tag:') and len(value) > len('$tag:'):
return True
if value.startswith('$type:') and len(value) > len('$type:'):
return True
if for_target and value.startswith('$dispvm:$tag:') and \
len(value) > len('$dispvm:$tag:'):
return True
if value == '$anyvm':
return True
if for_target and value == '$default':
return True
# those can be used to name one specific call VM
if value == '$adminvm':
return True
elif value.startswith('$type:') and len(value) > len('$type:'):
# allow only specific dispvm, not based on any $xxx keyword - don't name
# $tag here specifically, to work also with any future keywords
if for_target and value.startswith('$dispvm:') and \
not value.startswith('$dispvm:$'):
return True
elif value == '$anyvm':
return True
elif value == '$adminvm':
return True
elif value.startswith('$dispvm:') and for_target:
return True
elif value == '$dispvm' and for_target:
return True
elif value == '$default' and for_target:
if for_target and value == '$dispvm':
return True
return False
@ -179,12 +192,12 @@ class PolicyRule(object):
# verify special values
if self.source.startswith('$'):
if not verify_special_value(self.source, False):
if not verify_special_value(self.source, False, False):
raise PolicySyntaxError(filename, lineno,
'invalid source specification: {}'.format(self.source))
if self.target.startswith('$'):
if not verify_special_value(self.target, True):
if not verify_special_value(self.target, True, False):
raise PolicySyntaxError(filename, lineno,
'invalid target specification: {}'.format(self.target))
@ -196,8 +209,13 @@ class PolicyRule(object):
if self.override_target is not None:
if self.override_target.startswith('$') and \
not self.override_target.startswith('$dispvm') and \
self.override_target != '$adminvm':
not verify_special_value(self.override_target, True, True):
raise PolicySyntaxError(filename, lineno,
'target= option needs to name specific target')
if self.default_target is not None:
if self.default_target.startswith('$') and \
not verify_special_value(self.default_target, True, True):
raise PolicySyntaxError(filename, lineno,
'target= option needs to name specific target')
@ -244,9 +262,18 @@ class PolicyRule(object):
if value == policy_value:
return True
# if $dispvm* not matched above, reject it; missing ':' is
# intentional - handle both '$dispvm' and '$dispvm:xxx'
if value.startswith('$dispvm'):
# DispVM request, using tags to match
if policy_value.startswith('$dispvm:$tag:') \
and value.startswith('$dispvm:'):
tag = policy_value.split(':', 2)[2]
dispvm_base = value.split(':', 1)[1]
# already checked for existence by verify_target_value call
dispvm_base_info = system_info['domains'][dispvm_base]
return tag in dispvm_base_info['tags']
# if $dispvm* not matched above, reject it; default DispVM (bare
# $dispvm) was resolved by the caller
if value.startswith('$dispvm:'):
return False
# require $adminvm to be matched explicitly (not through $tag or $type)
@ -281,6 +308,17 @@ class PolicyRule(object):
if not self.is_match_single(system_info, self.source, source):
return False
# $dispvm in policy matches _only_ $dispvm (but not $dispvm:some-vm,
# even if that would be the default one)
if self.target == '$dispvm' and target == '$dispvm':
return True
if target == '$dispvm':
# resolve default DispVM, to check all kinds of $dispvm:*
default_dispvm = system_info['domains'][source]['default_dispvm']
if default_dispvm is None:
# if this VM have no default DispVM, match only with $anyvm
return self.target == '$anyvm'
target = '$dispvm:' + default_dispvm
if not self.is_match_single(system_info, self.target, target):
return False
return True
@ -307,13 +345,19 @@ class PolicyRule(object):
for name, domain in system_info['domains'].items():
if name != 'dom0':
yield name
if domain['dispvm_allowed']:
if domain['template_for_dispvms']:
yield '$dispvm:' + name
yield '$dispvm'
elif self.target.startswith('$dispvm:$tag:'):
tag = self.target.split(':', 2)[2]
for name, domain in system_info['domains'].items():
if tag in domain['tags']:
if domain['template_for_dispvms']:
yield '$dispvm:' + name
elif self.target.startswith('$dispvm:'):
dispvm_base = self.target.split(':', 1)[1]
try:
if system_info['domains'][dispvm_base]['dispvm_allowed']:
if system_info['domains'][dispvm_base]['template_for_dispvms']:
yield self.target
except KeyError:
# TODO log a warning?
@ -685,7 +729,7 @@ def get_system_info():
- `<domain name>`:
- tags: list of tags
- type: domain type
- dispvm_allowed: should DispVM based on this VM be allowed
- template_for_dispvms: should DispVM based on this VM be allowed
- default_dispvm: name of default AppVM for DispVMs started from here
'''

3
qubespolicy/cli.py
View File

@ -115,7 +115,8 @@ def main(args=None):
icons = {name: system_info['domains'][name]['icon']
for name in system_info['domains'].keys()}
for dispvm_base in system_info['domains']:
if not system_info['domains'][dispvm_base]['dispvm_allowed']:
if not (system_info['domains'][dispvm_base]
['template_for_dispvms']):
continue
dispvm_api_name = '$dispvm:' + dispvm_base
icons[dispvm_api_name] = \

2
qubespolicy/graph.py
View File

@ -87,7 +87,7 @@ def main(args=None):
targets = list(system_info['domains'].keys())
targets.append('$dispvm')
targets.extend('$dispvm:' + dom for dom in system_info['domains']
if system_info['domains'][dom]['dispvm_allowed'])
if system_info['domains'][dom]['template_for_dispvms'])
connections = set()

102
qubespolicy/tests/__init__.py
View File

@ -34,55 +34,55 @@ system_info = {
'tags': ['dom0-tag'],
'type': 'AdminVM',
'default_dispvm': 'default-dvm',
'dispvm_allowed': False,
'template_for_dispvms': False,
},
'test-vm1': {
'tags': ['tag1', 'tag2'],
'type': 'AppVM',
'default_dispvm': 'default-dvm',
'dispvm_allowed': False,
'template_for_dispvms': False,
},
'test-vm2': {
'tags': ['tag2'],
'type': 'AppVM',
'default_dispvm': 'default-dvm',
'dispvm_allowed': False,
'template_for_dispvms': False,
},
'test-vm3': {
'tags': [],
'tags': ['tag3'],
'type': 'AppVM',
'default_dispvm': 'default-dvm',
'dispvm_allowed': True,
'template_for_dispvms': True,
},
'default-dvm': {
'tags': [],
'type': 'AppVM',
'default_dispvm': 'default-dvm',
'dispvm_allowed': True,
'template_for_dispvms': True,
},
'test-invalid-dvm': {
'tags': ['tag1', 'tag2'],
'type': 'AppVM',
'default_dispvm': 'test-vm1',
'dispvm_allowed': False,
'template_for_dispvms': False,
},
'test-no-dvm': {
'tags': ['tag1', 'tag2'],
'type': 'AppVM',
'default_dispvm': None,
'dispvm_allowed': False,
'template_for_dispvms': False,
},
'test-template': {
'tags': ['tag1', 'tag2'],
'type': 'TemplateVM',
'default_dispvm': 'default-dvm',
'dispvm_allowed': False,
'template_for_dispvms': False,
},
'test-standalone': {
'tags': ['tag1', 'tag2'],
'type': 'StandaloneVM',
'default_dispvm': 'default-dvm',
'dispvm_allowed': False,
'template_for_dispvms': False,
},
}
}
@ -119,6 +119,8 @@ class TC_00_PolicyRule(qubes.tests.QubesTestCase):
qubespolicy.verify_target_value(system_info, '$anyvm'))
self.assertFalse(
qubespolicy.verify_target_value(system_info, '$tag:tag1'))
self.assertFalse(
qubespolicy.verify_target_value(system_info, '$dispvm:$tag:tag1'))
self.assertFalse(
qubespolicy.verify_target_value(system_info, '$invalid'))
@ -131,12 +133,18 @@ class TC_00_PolicyRule(qubes.tests.QubesTestCase):
for_target=False))
self.assertTrue(qubespolicy.verify_special_value('$adminvm',
for_target=False))
self.assertTrue(qubespolicy.verify_special_value('$dispvm:some-vm',
for_target=True))
self.assertTrue(qubespolicy.verify_special_value('$dispvm:$tag:tag1',
for_target=True))
self.assertFalse(qubespolicy.verify_special_value('$default',
for_target=False))
self.assertFalse(qubespolicy.verify_special_value('$dispvm',
for_target=False))
self.assertFalse(qubespolicy.verify_special_value('$dispvm:some-vm',
for_target=False))
self.assertFalse(qubespolicy.verify_special_value('$dispvm:$tag:tag1',
for_target=False))
self.assertFalse(qubespolicy.verify_special_value('$invalid',
for_target=False))
self.assertFalse(qubespolicy.verify_special_value('vm-name',
@ -219,6 +227,9 @@ class TC_00_PolicyRule(qubes.tests.QubesTestCase):
invalid_lines = [
'$dispvm $default allow', # $dispvm can't be a source
'$default $default allow', # $default can't be a source
'$anyvm $default allow,target=$dispvm:$tag:tag1', # $dispvm:$tag
# as override target
'$anyvm $default allow,target=$tag:tag1', # $tag as override target
'$anyvm $default deny,target=test-vm1', # target= used with deny
'$anyvm $anyvm deny,default_target=test-vm1', # default_target=
# with deny
@ -254,6 +265,8 @@ class TC_00_PolicyRule(qubes.tests.QubesTestCase):
self.assertTrue(is_match_single(system_info,
'$anyvm', '$dispvm:default-dvm'))
self.assertTrue(is_match_single(system_info, '$dispvm', '$dispvm'))
self.assertTrue(is_match_single(system_info,
'$dispvm:$tag:tag3', '$dispvm:test-vm3'))
self.assertTrue(is_match_single(system_info, '$adminvm', '$adminvm'))
self.assertTrue(is_match_single(system_info, '$adminvm', 'dom0'))
self.assertTrue(is_match_single(system_info, 'dom0', '$adminvm'))
@ -268,12 +281,20 @@ class TC_00_PolicyRule(qubes.tests.QubesTestCase):
self.assertFalse(is_match_single(system_info, '$default', 'test-vm1'))
self.assertFalse(is_match_single(system_info, '$tag:tag1', 'test-vm3'))
self.assertFalse(is_match_single(system_info, '$anyvm', 'no-such-vm'))
# test-vm1.dispvm_allowed=False
# test-vm1.template_for_dispvms=False
self.assertFalse(is_match_single(system_info,
'$anyvm', '$dispvm:test-vm1'))
# test-vm1.dispvm_allowed=False
# test-vm1.template_for_dispvms=False
self.assertFalse(is_match_single(system_info,
'$dispvm:test-vm1', '$dispvm:test-vm1'))
self.assertFalse(is_match_single(system_info,
'$dispvm:$tag:tag1', '$dispvm:test-vm1'))
# test-vm3 has not tag1
self.assertFalse(is_match_single(system_info,
'$dispvm:$tag:tag1', '$dispvm:test-vm3'))
# default-dvm has no tag3
self.assertFalse(is_match_single(system_info,
'$dispvm:$tag:tag3', '$dispvm:default-dvm'))
self.assertFalse(is_match_single(system_info, '$anyvm', 'dom0'))
self.assertFalse(is_match_single(system_info, '$anyvm', '$adminvm'))
self.assertFalse(is_match_single(system_info,
@ -306,6 +327,19 @@ class TC_00_PolicyRule(qubes.tests.QubesTestCase):
self.assertFalse(line.is_match(system_info, 'no-such-vm', 'test-vm2'))
line = qubespolicy.PolicyRule('$anyvm $anyvm allow')
self.assertFalse(line.is_match(system_info, 'test-vm1', 'no-such-vm'))
line = qubespolicy.PolicyRule('$anyvm $dispvm allow')
self.assertTrue(line.is_match(system_info, 'test-vm1', '$dispvm'))
line = qubespolicy.PolicyRule('$anyvm $dispvm allow')
self.assertFalse(line.is_match(system_info,
'test-vm1', '$dispvm:default-dvm'))
line = qubespolicy.PolicyRule('$anyvm $dispvm:default-dvm allow')
self.assertTrue(line.is_match(system_info, 'test-vm1', '$dispvm'))
line = qubespolicy.PolicyRule('$anyvm $dispvm:default-dvm allow')
self.assertTrue(line.is_match(system_info,
'test-vm1', '$dispvm:default-dvm'))
line = qubespolicy.PolicyRule('$anyvm $dispvm:$tag:tag3 allow')
self.assertTrue(line.is_match(system_info,
'test-vm1', '$dispvm:test-vm3'))
def test_060_expand_target(self):
lines = {
@ -317,6 +351,9 @@ class TC_00_PolicyRule(qubes.tests.QubesTestCase):
'$anyvm $dispvm:default-dvm allow': ['$dispvm:default-dvm'],
# no DispVM from test-vm1 allowed
'$anyvm $dispvm:test-vm1 allow': [],
'$anyvm $dispvm:test-vm3 allow': ['$dispvm:test-vm3'],
'$anyvm $dispvm:$tag:tag1 allow': [],
'$anyvm $dispvm:$tag:tag3 allow': ['$dispvm:test-vm3'],
'$anyvm test-vm1 allow': ['test-vm1'],
'$anyvm $type:AppVM allow': ['test-vm1', 'test-vm2', 'test-vm3',
'default-dvm', 'test-invalid-dvm', 'test-no-dvm'],
@ -599,6 +636,9 @@ class TC_20_Policy(qubes.tests.QubesTestCase):
f.write('test-vm1 $anyvm ask\n')
f.write('test-vm2 $tag:tag1 deny\n')
f.write('test-vm2 $tag:tag2 allow\n')
f.write('test-vm2 $dispvm:$tag:tag3 allow\n')
f.write('test-vm2 $dispvm:$tag:tag2 allow\n')
f.write('test-vm2 $dispvm:default-dvm allow\n')
f.write('$type:AppVM $default allow,target=test-vm3\n')
f.write('$tag:tag1 $type:AppVM allow\n')
policy = qubespolicy.Policy('test.service', tmp_policy_dir)
@ -614,14 +654,22 @@ class TC_20_Policy(qubes.tests.QubesTestCase):
self.assertEqual(policy.find_matching_rule(
system_info, 'test-vm1', ''), policy.policy_rules[1])
self.assertEqual(policy.find_matching_rule(
system_info, 'test-vm2', ''), policy.policy_rules[4])
system_info, 'test-vm2', ''), policy.policy_rules[7])
self.assertEqual(policy.find_matching_rule(
system_info, 'test-vm2', '$default'), policy.policy_rules[4])
system_info, 'test-vm2', '$default'), policy.policy_rules[7])
self.assertEqual(policy.find_matching_rule(
system_info, 'test-no-dvm', 'test-vm3'), policy.policy_rules[5])
system_info, 'test-no-dvm', 'test-vm3'), policy.policy_rules[8])
self.assertEqual(policy.find_matching_rule(
system_info, 'test-vm2', '$dispvm:test-vm3'),
policy.policy_rules[4])
self.assertEqual(policy.find_matching_rule(
system_info, 'test-vm2', '$dispvm'),
policy.policy_rules[6])
with self.assertRaises(qubespolicy.AccessDenied):
policy.find_matching_rule(
system_info, 'test-no-dvm', 'test-standalone')
with self.assertRaises(qubespolicy.AccessDenied):
policy.find_matching_rule(system_info, 'test-no-dvm', '$dispvm')
with self.assertRaises(qubespolicy.AccessDenied):
policy.find_matching_rule(
system_info, 'test-standalone', '$default')
@ -745,6 +793,30 @@ class TC_20_Policy(qubes.tests.QubesTestCase):
self.assertEqual(action.service, 'test.service')
self.assertIsNone(action.targets_for_ask)
def test_035_eval_resolve_dispvm_fail(self):
with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
f.write('test-no-dvm $dispvm allow\n')
policy = qubespolicy.Policy('test.service', tmp_policy_dir)
with self.assertRaises(qubespolicy.AccessDenied):
policy.evaluate(system_info, 'test-no-dvm', '$dispvm')
def test_036_eval_invalid_override_target(self):
with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
f.write('test-vm3 $anyvm allow,target=no-such-vm\n')
policy = qubespolicy.Policy('test.service', tmp_policy_dir)
with self.assertRaises(qubespolicy.AccessDenied):
policy.evaluate(system_info, 'test-vm3', '$default')
def test_037_eval_ask_no_targets(self):
with open(os.path.join(tmp_policy_dir, 'test.service'), 'w') as f:
f.write('test-vm3 $default ask\n')
policy = qubespolicy.Policy('test.service', tmp_policy_dir)
with self.assertRaises(qubespolicy.AccessDenied):
policy.evaluate(system_info, 'test-vm3', '$default')
class TC_30_Misc(qubes.tests.QubesTestCase):
@unittest.mock.patch('socket.socket')

8
qubespolicy/tests/cli.py
View File

@ -41,10 +41,10 @@ class TC_00_qrexec_policy(qubes.tests.QubesTestCase):
self.system_info_mock = self.system_info_patch.start()
self.system_info = {
'domains': {'dom0': {'icon': 'black', 'dispvm_allowed': False},
'test-vm1': {'icon': 'red', 'dispvm_allowed': False},
'test-vm2': {'icon': 'red', 'dispvm_allowed': False},
'test-vm3': {'icon': 'green', 'dispvm_allowed': True}, }}
'domains': {'dom0': {'icon': 'black', 'template_for_dispvms': False},
'test-vm1': {'icon': 'red', 'template_for_dispvms': False},
'test-vm2': {'icon': 'red', 'template_for_dispvms': False},
'test-vm3': {'icon': 'green', 'template_for_dispvms': True}, }}
self.system_info_mock.return_value = self.system_info
self.dbus_patch = unittest.mock.patch('pydbus.SystemBus')