Merge remote-tracking branches 'origin/pull/142/head', 'origin/pull/143/head' and 'origin/pull/144/head'

This commit is contained in:
Wojtek Porczyk 2017-08-14 16:32:56 +02:00
commit 044e10a6ec
14 changed files with 647 additions and 16 deletions

View File

@ -173,6 +173,8 @@ endif
cp qubes-rpc/qubes.NotifyTools $(DESTDIR)/etc/qubes-rpc/
cp qubes-rpc/qubes.NotifyUpdates $(DESTDIR)/etc/qubes-rpc/
install qubes-rpc/qubesd-query-fast $(DESTDIR)/usr/libexec/qubes/
install -m 0755 qvm-tools/qubes-bug-report $(DESTDIR)/usr/bin/qubes-bug-report
install -m 0755 qvm-tools/qubes-hcl-report $(DESTDIR)/usr/bin/qubes-hcl-report
install -m 0755 qvm-tools/qvm-sync-clock $(DESTDIR)/usr/bin/qvm-sync-clock
for method in $(ADMIN_API_METHODS_SIMPLE); do \
ln -s ../../usr/libexec/qubes/qubesd-query-fast \

View File

@ -28,8 +28,9 @@ class ServicesExtension(qubes.ext.Extension):
'''
# pylint: disable=no-self-use
@qubes.ext.handler('domain-qdb-create')
def on_domain_qdb_create(self, vm):
def on_domain_qdb_create(self, vm, event):
'''Actually export features'''
# pylint: disable=unused-argument
for feature, value in vm.features.items():
if not feature.startswith('service.'):
continue
@ -39,7 +40,7 @@ class ServicesExtension(qubes.ext.Extension):
str(int(bool(value))))
@qubes.ext.handler('domain-feature-set')
def on_domain_feature_set(self, vm, feature, value, oldvalue=None):
def on_domain_feature_set(self, vm, event, feature, value, oldvalue=None):
'''Update /qubes-service/ QubesDB tree in runtime'''
# pylint: disable=unused-argument
if not vm.is_running():
@ -52,8 +53,9 @@ class ServicesExtension(qubes.ext.Extension):
str(int(bool(value))))
@qubes.ext.handler('domain-feature-delete')
def on_domain_feature_delete(self, vm, feature):
def on_domain_feature_delete(self, vm, event, feature):
'''Update /qubes-service/ QubesDB tree in runtime'''
# pylint: disable=unused-argument
if not vm.is_running():
return
if not feature.startswith('service.'):

View File

@ -1010,6 +1010,7 @@ def load_tests(loader, tests, pattern): # pylint: disable=unused-argument
'qubes.tests.api_admin',
'qubes.tests.api_misc',
'qubespolicy.tests',
'qubespolicy.tests.cli',
):
tests.addTests(loader.loadTestsFromName(modname))

View File

@ -60,6 +60,9 @@ def validate_name(holder, prop, value):
else:
raise qubes.exc.QubesValueError(
'VM name contains illegal characters')
if value in ('none', 'default'):
raise qubes.exc.QubesValueError(
'VM name cannot be \'none\' nor \'default\'')
class Features(dict):

View File

@ -47,6 +47,12 @@ class PolicySyntaxError(AccessDenied):
super(PolicySyntaxError, self).__init__(
'{}:{}: {}'.format(filename, lineno, msg))
class PolicyNotFound(AccessDenied):
''' Policy was not found for this service '''
def __init__(self, service_name):
super(PolicyNotFound, self).__init__(
'Policy not found for service {}'.format(service_name))
class Action(enum.Enum):
''' Action as defined by policy '''
@ -486,6 +492,8 @@ class Policy(object):
if not os.path.exists(policy_file):
# fallback to policy without specific argument set (if any)
policy_file = os.path.join(policy_dir, service.split('+')[0])
if not os.path.exists(policy_file):
raise PolicyNotFound(service)
#: policy storage directory
self.policy_dir = policy_dir

View File

@ -30,6 +30,7 @@ from gi.repository import GLib
# pylint: enable=import-error
import qubespolicy.rpcconfirmation
import qubespolicy.policycreateconfirmation
# pylint: enable=wrong-import-position
class PolicyAgent(object):
@ -45,6 +46,11 @@ class PolicyAgent(object):
<arg type='a{ss}' name='icons' direction='in'/>
<arg type='s' name='response' direction='out'/>
</method>
<method name='ConfirmPolicyCreate'>
<arg type='s' name='source' direction='in'/>
<arg type='s' name='service_name' direction='in'/>
<arg type='b' name='response' direction='out'/>
</method>
</interface>
</node>
"""
@ -63,6 +69,13 @@ class PolicyAgent(object):
targets, default_target or None)
return response or ''
@staticmethod
def ConfirmPolicyCreate(source, service_name):
# pylint: disable=invalid-name
response = qubespolicy.policycreateconfirmation.confirm(
source, service_name)
return response
def main():
loop = GLib.MainLoop()

View File

@ -20,6 +20,7 @@
import argparse
import logging
import logging.handlers
import os
import sys
@ -46,6 +47,20 @@ parser.add_argument('process_ident', metavar='process-ident',
help='Qrexec process identifier - for connecting data channel')
def create_default_policy(service_name):
policy_file = os.path.join(qubespolicy.POLICY_DIR, service_name)
with open(policy_file, "w") as policy:
policy.write(
"## Policy file automatically created on first service call.\n")
policy.write(
"## Fill free to edit.\n")
policy.write("## Note that policy parsing stops at the first match\n")
policy.write("\n")
policy.write("## Please use a single # to start your custom comments\n")
policy.write("\n")
policy.write("$anyvm $anyvm ask\n")
def main(args=None):
args = parser.parse_args(args)
@ -64,7 +79,22 @@ def main(args=None):
log.error(log_prefix + 'error getting system info: ' + str(e))
return 1
try:
policy = qubespolicy.Policy(args.service_name)
try:
policy = qubespolicy.Policy(args.service_name)
except qubespolicy.PolicyNotFound:
service_name = args.service_name.split('+')[0]
import pydbus
bus = pydbus.SystemBus()
proxy = bus.get('org.qubesos.PolicyAgent',
'/org/qubesos/PolicyAgent')
create_policy = proxy.ConfirmPolicyCreate(
args.domain, service_name)
if create_policy:
create_default_policy(service_name)
policy = qubespolicy.Policy(args.service_name)
else:
raise
action = policy.evaluate(system_info, args.domain, args.target)
if args.assume_yes_for_ask and action.action == qubespolicy.Action.ask:
action.action = qubespolicy.Action.allow

View File

@ -0,0 +1,141 @@
<?xml version="1.0" encoding="UTF-8"?>
<!-- Generated with glade 3.20.0 -->
<interface>
<requires lib="gtk+" version="3.20"/>
<object class="GtkMessageDialog" id="PolicyCreateConfirmationWindow">
<property name="can_focus">False</property>
<property name="title" translatable="yes">Default service policy</property>
<property name="icon_name">dialog-warning</property>
<property name="type_hint">dialog</property>
<property name="message_type">question</property>
<property name="buttons">ok-cancel</property>
<child internal-child="vbox">
<object class="GtkBox">
<property name="can_focus">False</property>
<property name="orientation">vertical</property>
<child internal-child="action_area">
<object class="GtkButtonBox">
<property name="can_focus">False</property>
<property name="layout_style">end</property>
</object>
<packing>
<property name="expand">False</property>
<property name="fill">True</property>
<property name="position">0</property>
</packing>
</child>
<child>
<object class="GtkBox">
<property name="visible">True</property>
<property name="can_focus">False</property>
<property name="orientation">vertical</property>
<property name="spacing">2</property>
<child>
<object class="GtkLabel" id="messageLabel">
<property name="visible">True</property>
<property name="can_focus">False</property>
<property name="label" translatable="yes">Policy for requested service does not exist.
Do you want to create default one (ask for everything)?</property>
</object>
<packing>
<property name="expand">False</property>
<property name="fill">True</property>
<property name="padding">2</property>
<property name="position">0</property>
</packing>
</child>
<child>
<object class="GtkGrid">
<property name="visible">True</property>
<property name="can_focus">False</property>
<property name="row_spacing">2</property>
<property name="column_spacing">2</property>
<child>
<object class="GtkLabel" id="sourceLabel">
<property name="visible">True</property>
<property name="can_focus">False</property>
<property name="margin_left">4</property>
<property name="label" translatable="yes">Source:</property>
</object>
<packing>
<property name="left_attach">0</property>
<property name="top_attach">0</property>
</packing>
</child>
<child>
<object class="GtkLabel" id="serviceLabel">
<property name="visible">True</property>
<property name="can_focus">False</property>
<property name="margin_left">4</property>
<property name="label" translatable="yes">Service:</property>
</object>
<packing>
<property name="left_attach">0</property>
<property name="top_attach">1</property>
</packing>
</child>
<child>
<object class="GtkEntry" id="sourceEntry">
<property name="visible">True</property>
<property name="sensitive">False</property>
<property name="can_focus">True</property>
<property name="editable">False</property>
</object>
<packing>
<property name="left_attach">1</property>
<property name="top_attach">0</property>
</packing>
</child>
<child>
<object class="GtkEntry" id="serviceEntry">
<property name="visible">True</property>
<property name="sensitive">False</property>
<property name="can_focus">True</property>
<property name="editable">False</property>
</object>
<packing>
<property name="left_attach">1</property>
<property name="top_attach">1</property>
</packing>
</child>
<child>
<object class="GtkLabel" id="confirmLabel">
<property name="visible">True</property>
<property name="can_focus">False</property>
<property name="label" translatable="yes">Type capital YES to confirm: </property>
</object>
<packing>
<property name="left_attach">0</property>
<property name="top_attach">2</property>
</packing>
</child>
<child>
<object class="GtkEntry" id="confirmEntry">
<property name="visible">True</property>
<property name="can_focus">True</property>
<property name="activates_default">True</property>
<property name="caps_lock_warning">False</property>
</object>
<packing>
<property name="left_attach">1</property>
<property name="top_attach">2</property>
</packing>
</child>
</object>
<packing>
<property name="expand">False</property>
<property name="fill">False</property>
<property name="position">1</property>
</packing>
</child>
</object>
<packing>
<property name="expand">False</property>
<property name="fill">True</property>
<property name="position">1</property>
</packing>
</child>
</object>
</child>
</object>
</interface>

View File

@ -52,17 +52,19 @@ def handle_single_action(args, action):
service = ''
else:
service = action.service
target = action.target or action.original_target
# handle forced target=
if action.rule.override_target:
target = action.rule.override_target
if args.target and target not in args.target:
return ''
if action.action == qubespolicy.Action.ask:
if args.include_ask:
# handle forced target=
if len(action.targets_for_ask) == 1:
return ' "{}" -> "{}" [label="{}" color=orange];\n'.format(
action.source, action.targets_for_ask[0], service)
return ' "{}" -> "{}" [label="{}" color=orange];\n'.format(
action.source, action.original_target, service)
action.source, target, service)
elif action.action == qubespolicy.Action.allow:
return ' "{}" -> "{}" [label="{}" color=red];\n'.format(
action.source, action.target, service)
action.source, target, service)
return ''
def main(args=None):
@ -83,12 +85,9 @@ def main(args=None):
sources = args.source
targets = list(system_info['domains'].keys())
if args.target:
targets = args.target
else:
targets.append('$dispvm')
targets.extend('$dispvm:' + dom for dom in system_info['domains']
if system_info['domains'][dom]['dispvm_allowed'])
targets.append('$dispvm')
targets.extend('$dispvm:' + dom for dom in system_info['domains']
if system_info['domains'][dom]['dispvm_allowed'])
connections = set()

View File

@ -0,0 +1,82 @@
# -*- encoding: utf-8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2017 Marek Marczykowski-Górecki
# <marmarek@invisiblethingslab.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, see <http://www.gnu.org/licenses/>.
import os
import pkg_resources
# pylint: disable=import-error,wrong-import-position
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# pylint: enable=import-error
class PolicyCreateConfirmationWindow(object):
# pylint: disable=too-few-public-methods
_source_file = pkg_resources.resource_filename('qubespolicy',
os.path.join('glade', "PolicyCreateConfirmationWindow.glade"))
_source_id = {'window': "PolicyCreateConfirmationWindow",
'ok': "okButton",
'cancel': "cancelButton",
'source': "sourceEntry",
'service': "serviceEntry",
'confirm': "confirmEntry",
}
def __init__(self, source, service):
self._gtk_builder = Gtk.Builder()
self._gtk_builder.add_from_file(self._source_file)
self._window = self._gtk_builder.get_object(
self._source_id['window'])
self._rpc_ok_button = self._gtk_builder.get_object(
self._source_id['ok'])
self._rpc_cancel_button = self._gtk_builder.get_object(
self._source_id['cancel'])
self._service_entry = self._gtk_builder.get_object(
self._source_id['service'])
self._source_entry = self._gtk_builder.get_object(
self._source_id['source'])
self._confirm_entry = self._gtk_builder.get_object(
self._source_id['confirm'])
self._source_entry.set_text(source)
self._service_entry.set_text(service)
# make OK button the default
ok_button = self._window.get_widget_for_response(Gtk.ResponseType.OK)
ok_button.set_can_default(True)
ok_button.grab_default()
def run(self):
self._window.set_keep_above(True)
self._window.connect("delete-event", Gtk.main_quit)
self._window.show_all()
response = self._window.run()
self._window.hide()
if response == Gtk.ResponseType.OK:
return self._confirm_entry.get_text() == 'YES'
return False
def confirm(source, service):
window = PolicyCreateConfirmationWindow(source, service)
return window.run()

343
qubespolicy/tests/cli.py Normal file
View File

@ -0,0 +1,343 @@
# -*- encoding: utf-8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2017 Marek Marczykowski-Górecki
# <marmarek@invisiblethingslab.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, see <http://www.gnu.org/licenses/>.
import os
import tempfile
import unittest.mock
import shutil
import qubes.tests
import qubespolicy
import qubespolicy.cli
import qubespolicy.tests
class TC_00_qrexec_policy(qubes.tests.QubesTestCase):
def setUp(self):
super(TC_00_qrexec_policy, self).setUp()
self.policy_patch = unittest.mock.patch('qubespolicy.Policy')
self.policy_mock = self.policy_patch.start()
self.system_info_patch = unittest.mock.patch(
'qubespolicy.get_system_info')
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}, }}
self.system_info_mock.return_value = self.system_info
self.dbus_patch = unittest.mock.patch('pydbus.SystemBus')
self.dbus_mock = self.dbus_patch.start()
self.policy_dir = tempfile.TemporaryDirectory()
self.policydir_patch = unittest.mock.patch('qubespolicy.POLICY_DIR',
self.policy_dir.name)
self.policydir_patch.start()
def tearDown(self):
self.policydir_patch.stop()
self.policy_dir.cleanup()
self.dbus_patch.start()
self.system_info_patch.stop()
self.policy_patch.stop()
super(TC_00_qrexec_policy, self).tearDown()
def test_000_allow(self):
self.policy_mock.configure_mock(**{
'return_value.evaluate.return_value.action':
qubespolicy.Action.allow,
})
retval = qubespolicy.cli.main(
['source-id', 'source', 'target', 'service', 'process_ident'])
self.assertEqual(retval, 0)
self.assertEqual(self.policy_mock.mock_calls, [
('', ('service',), {}),
('().evaluate', (self.system_info, 'source',
'target'), {}),
('().evaluate().target.__str__', (), {}),
('().evaluate().execute', ('process_ident,source,source-id', ), {}),
])
self.assertEqual(self.dbus_mock.mock_calls, [])
def test_010_ask_allow(self):
self.policy_mock.configure_mock(**{
'return_value.evaluate.return_value.action':
qubespolicy.Action.ask,
'return_value.evaluate.return_value.target':
None,
'return_value.evaluate.return_value.targets_for_ask':
['test-vm1', 'test-vm2'],
})
self.dbus_mock.configure_mock(**{
'return_value.get.return_value.Ask.return_value': 'test-vm1'
})
retval = qubespolicy.cli.main(
['source-id', 'source', 'target', 'service', 'process_ident'])
self.assertEqual(retval, 0)
self.assertEqual(self.policy_mock.mock_calls, [
('', ('service',), {}),
('().evaluate', (self.system_info, 'source',
'target'), {}),
('().evaluate().handle_user_response', (True, 'test-vm1'), {}),
('().evaluate().execute', ('process_ident,source,source-id', ), {}),
])
icons = {
'dom0': 'black',
'test-vm1': 'red',
'test-vm2': 'red',
'test-vm3': 'green',
'$dispvm:test-vm3': 'green',
}
self.assertEqual(self.dbus_mock.mock_calls, [
('', (), {}),
('().get', ('org.qubesos.PolicyAgent',
'/org/qubesos/PolicyAgent'), {}),
('().get().Ask', ('source', 'service', ['test-vm1', 'test-vm2'],
'', icons), {}),
])
def test_011_ask_deny(self):
self.policy_mock.configure_mock(**{
'return_value.evaluate.return_value.action':
qubespolicy.Action.ask,
'return_value.evaluate.return_value.target':
None,
'return_value.evaluate.return_value.targets_for_ask':
['test-vm1', 'test-vm2'],
'return_value.evaluate.return_value.handle_user_response'
'.side_effect':
qubespolicy.AccessDenied,
})
self.dbus_mock.configure_mock(**{
'return_value.get.return_value.Ask.return_value': ''
})
retval = qubespolicy.cli.main(
['source-id', 'source', 'target', 'service', 'process_ident'])
self.assertEqual(retval, 1)
self.assertEqual(self.policy_mock.mock_calls, [
('', ('service',), {}),
('().evaluate', (self.system_info, 'source',
'target'), {}),
('().evaluate().handle_user_response', (False,), {}),
])
icons = {
'dom0': 'black',
'test-vm1': 'red',
'test-vm2': 'red',
'test-vm3': 'green',
'$dispvm:test-vm3': 'green',
}
self.assertEqual(self.dbus_mock.mock_calls, [
('', (), {}),
('().get', ('org.qubesos.PolicyAgent',
'/org/qubesos/PolicyAgent'), {}),
('().get().Ask', ('source', 'service', ['test-vm1', 'test-vm2'],
'', icons), {}),
])
def test_012_ask_default_target(self):
self.policy_mock.configure_mock(**{
'return_value.evaluate.return_value.action':
qubespolicy.Action.ask,
'return_value.evaluate.return_value.target':
'test-vm1',
'return_value.evaluate.return_value.targets_for_ask':
['test-vm1', 'test-vm2'],
})
self.dbus_mock.configure_mock(**{
'return_value.get.return_value.Ask.return_value': 'test-vm1'
})
retval = qubespolicy.cli.main(
['source-id', 'source', 'target', 'service', 'process_ident'])
self.assertEqual(retval, 0)
self.assertEqual(self.policy_mock.mock_calls, [
('', ('service',), {}),
('().evaluate', (self.system_info, 'source',
'target'), {}),
('().evaluate().handle_user_response', (True, 'test-vm1'), {}),
('().evaluate().execute', ('process_ident,source,source-id',), {}),
])
icons = {
'dom0': 'black',
'test-vm1': 'red',
'test-vm2': 'red',
'test-vm3': 'green',
'$dispvm:test-vm3': 'green',
}
self.assertEqual(self.dbus_mock.mock_calls, [
('', (), {}),
('().get', ('org.qubesos.PolicyAgent',
'/org/qubesos/PolicyAgent'), {}),
('().get().Ask', ('source', 'service', ['test-vm1', 'test-vm2'],
'test-vm1', icons), {}),
])
def test_020_deny(self):
self.policy_mock.configure_mock(**{
'return_value.evaluate.return_value.action':
qubespolicy.Action.deny,
'return_value.evaluate.return_value.execute.side_effect':
qubespolicy.AccessDenied,
})
retval = qubespolicy.cli.main(
['source-id', 'source', 'target', 'service', 'process_ident'])
self.assertEqual(retval, 1)
self.assertEqual(self.policy_mock.mock_calls, [
('', ('service',), {}),
('().evaluate', (self.system_info, 'source',
'target'), {}),
('().evaluate().target.__str__', (), {}),
('().evaluate().execute', ('process_ident,source,source-id',), {}),
])
self.assertEqual(self.dbus_mock.mock_calls, [])
def test_030_just_evaluate_allow(self):
self.policy_mock.configure_mock(**{
'return_value.evaluate.return_value.action':
qubespolicy.Action.allow,
})
retval = qubespolicy.cli.main(
['--just-evaluate',
'source-id', 'source', 'target', 'service', 'process_ident'])
self.assertEqual(retval, 0)
self.assertEqual(self.policy_mock.mock_calls, [
('', ('service',), {}),
('().evaluate', (self.system_info, 'source',
'target'), {}),
])
self.assertEqual(self.dbus_mock.mock_calls, [])
def test_031_just_evaluate_deny(self):
self.policy_mock.configure_mock(**{
'return_value.evaluate.return_value.action':
qubespolicy.Action.deny,
})
retval = qubespolicy.cli.main(
['--just-evaluate',
'source-id', 'source', 'target', 'service', 'process_ident'])
self.assertEqual(retval, 1)
self.assertEqual(self.policy_mock.mock_calls, [
('', ('service',), {}),
('().evaluate', (self.system_info, 'source',
'target'), {}),
])
self.assertEqual(self.dbus_mock.mock_calls, [])
def test_032_just_evaluate_ask(self):
self.policy_mock.configure_mock(**{
'return_value.evaluate.return_value.action':
qubespolicy.Action.ask,
})
retval = qubespolicy.cli.main(
['--just-evaluate',
'source-id', 'source', 'target', 'service', 'process_ident'])
self.assertEqual(retval, 1)
self.assertEqual(self.policy_mock.mock_calls, [
('', ('service',), {}),
('().evaluate', (self.system_info, 'source',
'target'), {}),
])
self.assertEqual(self.dbus_mock.mock_calls, [])
def test_033_just_evaluate_ask_assume_yes(self):
self.policy_mock.configure_mock(**{
'return_value.evaluate.return_value.action':
qubespolicy.Action.ask,
})
retval = qubespolicy.cli.main(
['--just-evaluate', '--assume-yes-for-ask',
'source-id', 'source', 'target', 'service', 'process_ident'])
self.assertEqual(retval, 0)
self.assertEqual(self.policy_mock.mock_calls, [
('', ('service',), {}),
('().evaluate', (self.system_info, 'source',
'target'), {}),
])
self.assertEqual(self.dbus_mock.mock_calls, [])
def test_040_create_policy(self):
self.policy_mock.configure_mock(**{
'side_effect':
[qubespolicy.PolicyNotFound('service'), unittest.mock.DEFAULT],
'return_value.evaluate.return_value.action':
qubespolicy.Action.allow,
})
self.dbus_mock.configure_mock(**{
'return_value.get.return_value.ConfirmPolicyCreate.return_value':
True
})
retval = qubespolicy.cli.main(
['source-id', 'source', 'target', 'service', 'process_ident'])
self.assertEqual(retval, 0)
self.assertEqual(self.policy_mock.mock_calls, [
('', ('service',), {}),
('', ('service',), {}),
('().evaluate', (self.system_info, 'source',
'target'), {}),
('().evaluate().target.__str__', (), {}),
('().evaluate().execute', ('process_ident,source,source-id',), {}),
])
self.assertEqual(self.dbus_mock.mock_calls, [
('', (), {}),
('().get', ('org.qubesos.PolicyAgent',
'/org/qubesos/PolicyAgent'), {}),
('().get().ConfirmPolicyCreate', ('source', 'service'), {}),
])
policy_path = os.path.join(self.policy_dir.name, 'service')
self.assertTrue(os.path.exists(policy_path))
with open(policy_path) as policy_file:
self.assertEqual(policy_file.read(),
"## Policy file automatically created on first service call.\n"
"## Fill free to edit.\n"
"## Note that policy parsing stops at the first match\n"
"\n"
"## Please use a single # to start your custom comments\n"
"\n"
"$anyvm $anyvm ask\n")
def test_041_create_policy_abort(self):
self.policy_mock.configure_mock(**{
'side_effect':
[qubespolicy.PolicyNotFound('service'), unittest.mock.DEFAULT],
'return_value.evaluate.return_value.action':
qubespolicy.Action.deny,
})
self.dbus_mock.configure_mock(**{
'return_value.get.return_value.ConfirmPolicyCreate.return_value':
False
})
retval = qubespolicy.cli.main(
['source-id', 'source', 'target', 'service', 'process_ident'])
self.assertEqual(retval, 1)
self.assertEqual(self.policy_mock.mock_calls, [
('', ('service',), {}),
])
self.assertEqual(self.dbus_mock.mock_calls, [
('', (), {}),
('().get', ('org.qubesos.PolicyAgent',
'/org/qubesos/PolicyAgent'), {}),
('().get().ConfirmPolicyCreate', ('source', 'service'), {}),
])
policy_path = os.path.join(self.policy_dir.name, 'service')
self.assertFalse(os.path.exists(policy_path))

View File

@ -374,6 +374,7 @@ fi
%{python3_sitelib}/qubespolicy/cli.py
%{python3_sitelib}/qubespolicy/agent.py
%{python3_sitelib}/qubespolicy/gtkhelpers.py
%{python3_sitelib}/qubespolicy/policycreateconfirmation.py
%{python3_sitelib}/qubespolicy/rpcconfirmation.py
%{python3_sitelib}/qubespolicy/utils.py
%{python3_sitelib}/qubespolicy/graph.py
@ -382,10 +383,12 @@ fi
%dir %{python3_sitelib}/qubespolicy/tests/__pycache__
%{python3_sitelib}/qubespolicy/tests/__pycache__/*
%{python3_sitelib}/qubespolicy/tests/__init__.py
%{python3_sitelib}/qubespolicy/tests/cli.py
%{python3_sitelib}/qubespolicy/tests/gtkhelpers.py
%{python3_sitelib}/qubespolicy/tests/rpcconfirmation.py
%dir %{python3_sitelib}/qubespolicy/glade
%{python3_sitelib}/qubespolicy/glade/PolicyCreateConfirmationWindow.glade
%{python3_sitelib}/qubespolicy/glade/RPCConfirmationWindow.glade
/usr/lib/qubes/cleanup-dispvms
@ -414,6 +417,7 @@ fi
/etc/xen/scripts/block-snapshot
/etc/xen/scripts/block-origin
/etc/xen/scripts/vif-route-qubes
%attr(2775,root,qubes) %dir /etc/qubes-rpc/policy
%attr(0664,root,qubes) %config(noreplace) /etc/qubes-rpc/policy/admin.*
%attr(0664,root,qubes) %config(noreplace) /etc/qubes-rpc/policy/include/admin-local-ro
%attr(0664,root,qubes) %config(noreplace) /etc/qubes-rpc/policy/include/admin-local-rwx

View File

@ -51,6 +51,7 @@ if __name__ == '__main__':
'qubes.ext.r3compatibility = qubes.ext.r3compatibility:R3Compatibility',
'qubes.ext.pci = qubes.ext.pci:PCIDeviceExtension',
'qubes.ext.block = qubes.ext.block:BlockDeviceExtension',
'qubes.ext.services = qubes.ext.services:ServicesExtension',
],
'qubes.devices': [
'pci = qubes.ext.pci:PCIDevice',

2
test-packages/pydbus.py Normal file
View File

@ -0,0 +1,2 @@
class SystemBus(object):
pass