Remove tools that are moved to -client repository

QubesOS/qubes-issues#853
This commit is contained in:
Marek Marczykowski-Górecki 2017-05-11 21:24:20 +02:00
parent 8992e71f85
commit 83eef56f9d
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
31 changed files with 1 additions and 3193 deletions

View File

@ -55,7 +55,6 @@ system_path = {
'qubes_pciback_cmd': '/usr/lib/qubes/unbind-pci-device.sh',
'prepare_volatile_img_cmd': '/usr/lib/qubes/prepare-volatile-img.sh',
'monitor_layout_notify_cmd': '/usr/bin/qubes-monitor-layout-notify',
}
vm_files = {

View File

@ -906,8 +906,6 @@ def load_tests(loader, tests, pattern): # pylint: disable=unused-argument
'qubes.tests.app',
'qubes.tests.tarwriter',
'qubes.tests.mgmt',
'qubes.tests.tools.qvm_device',
'qubes.tests.tools.qvm_firewall',
'qubespolicy.tests',
'qubes.tests.tools.qubesd',
):
@ -940,14 +938,6 @@ def load_tests(loader, tests, pattern): # pylint: disable=unused-argument
'qubes.tests.integ.backupcompatibility',
# 'qubes.tests.regressions',
# tool tests
'qubes.tests.integ.tools.qubes_create',
'qubes.tests.integ.tools.qvm_check',
'qubes.tests.integ.tools.qvm_features',
'qubes.tests.integ.tools.qvm_firewall',
'qubes.tests.integ.tools.qvm_prefs',
'qubes.tests.integ.tools.qvm_run',
# external modules
# 'qubes.tests.extra',
):

View File

@ -1,72 +0,0 @@
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2016 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import qubes
import qubes.tools.qvm_check
import qubes.tests
import qubes.vm.appvm
class TC_00_qvm_check(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
def setUp(self):
super(TC_00_qvm_check, self).setUp()
self.init_default_template()
self.sharedopts = ['--qubesxml', qubes.tests.XMLPATH]
self.vm1 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('vm1'),
template=self.app.default_template,
label='red')
self.vm1.create_on_disk()
self.app.save()
def test_000_exists(self):
self.assertEqual(0, qubes.tools.qvm_check.main(
self.sharedopts + [self.vm1.name]))
with self.assertRaises(SystemExit):
qubes.tools.qvm_check.main(
self.sharedopts + ['test-no-such-vm'])
def test_001_running(self):
self.assertEqual(1, qubes.tools.qvm_check.main(
self.sharedopts + ['--running', self.vm1.name]))
self.vm1.start()
self.assertEqual(0, qubes.tools.qvm_check.main(
self.sharedopts + ['--running', self.vm1.name]))
def test_002_paused(self):
self.assertEqual(1, qubes.tools.qvm_check.main(
self.sharedopts + ['--paused', self.vm1.name]))
self.vm1.start()
self.assertEqual(1, qubes.tools.qvm_check.main(
self.sharedopts + ['--paused', self.vm1.name]))
self.vm1.pause()
self.assertEqual(0, qubes.tools.qvm_check.main(
self.sharedopts + ['--paused', self.vm1.name]))
def test_003_template(self):
self.assertEqual(1, qubes.tools.qvm_check.main(
self.sharedopts + ['--template', self.vm1.name]))
self.assertEqual(0, qubes.tools.qvm_check.main(
self.sharedopts + ['--template', self.app.default_template.name]))

View File

@ -1,70 +0,0 @@
#!/usr/bin/python2
# -*- encoding: utf8 -*-
#
# 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
import qubes
import qubes.tools.qvm_features
import qubes.tests
import qubes.tests.tools
import qubes.vm.appvm
class TC_00_qvm_features(qubes.tests.SystemTestsMixin,
qubes.tests.QubesTestCase):
def setUp(self):
super(TC_00_qvm_features, self).setUp()
self.init_default_template()
self.sharedopts = ['--qubesxml', qubes.tests.XMLPATH]
self.vm1 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('vm1'),
template=self.app.default_template,
label='red')
self.app.save()
def test_000_list(self):
self.assertEqual(0, qubes.tools.qvm_features.main(
self.sharedopts + [self.vm1.name]))
with self.assertRaises(SystemExit):
qubes.tools.qvm_features.main(
self.sharedopts + ['test-no-such-vm'])
def test_001_get_missing(self):
self.assertEqual(1, qubes.tools.qvm_features.main(
self.sharedopts + [self.vm1.name, 'no-such-feature']))
def test_002_set_and_get(self):
self.assertEqual(0, qubes.tools.qvm_features.main(
self.sharedopts + [self.vm1.name, 'test-feature', 'true']))
with qubes.tests.tools.StdoutBuffer() as buf:
self.assertEqual(0, qubes.tools.qvm_features.main(
self.sharedopts + [self.vm1.name, 'test-feature']))
self.assertEqual('true\n', buf.getvalue())
def test_003_set_and_list(self):
self.assertEqual(0, qubes.tools.qvm_features.main(
self.sharedopts + [self.vm1.name, 'test-feature', 'true']))
with qubes.tests.tools.StdoutBuffer() as buf:
self.assertEqual(0, qubes.tools.qvm_features.main(
self.sharedopts + [self.vm1.name]))
self.assertEqual('test-feature true\n', buf.getvalue())

View File

@ -1,160 +0,0 @@
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2016 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import qubes.firewall
import qubes.tests
import qubes.tests.tools
import qubes.tools.qvm_firewall
import qubes.vm.appvm
class TC_10_ArgParser(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
list_header = ['NO', 'ACTION', 'HOST', 'PROTOCOL', 'PORT(S)',
'SPECIAL TARGET', 'ICMP TYPE']
def setUp(self):
super(TC_10_ArgParser, self).setUp()
self.init_default_template()
self.vm = self.app.add_new_vm(qubes.vm.appvm.AppVM, None,
name=self.make_vm_name('vm'), label='red')
self.vm.create_on_disk()
self.app.save()
def test_000_list(self):
with qubes.tests.tools.StdoutBuffer() as stdout:
qubes.tools.qvm_firewall.main([self.vm.name, 'list'])
self.assertEqual(stdout.getvalue(),
' '.join(self.list_header) + '\n')
def test_001_list(self):
self.vm.firewall.rules.append(
qubes.firewall.Rule(action='accept', dsthost='127.0.0.2',
proto='tcp', dstports=80))
self.vm.firewall.rules.append(
qubes.firewall.Rule(action='accept', dsthost='127.0.0.3',
proto='icmp', icmptype=8))
self.vm.firewall.rules.append(
qubes.firewall.Rule(action='accept', specialtarget='dns'))
self.vm.firewall.save()
expected_output = (
'NO ACTION HOST PROTOCOL PORT(S) SPECIAL TARGET ICMP '
'TYPE\n'
'0 accept 127.0.0.2/32 tcp 80 '
' \n'
'1 accept 127.0.0.3/32 icmp 8 '
' \n'
'2 accept dns '
' \n'
)
with qubes.tests.tools.StdoutBuffer() as stdout:
qubes.tools.qvm_firewall.main([self.vm.name, 'list'])
self.assertEqual(
'\n'.join(l.rstrip() for l in stdout.getvalue().splitlines()),
'\n'.join(l.rstrip() for l in expected_output.splitlines()))
def test_002_list_raw(self):
self.vm.firewall.rules = [
qubes.firewall.Rule(action='accept', dsthost='127.0.0.2',
proto='tcp', dstports=80),
qubes.firewall.Rule(action='accept', dsthost='127.0.0.3',
proto='icmp', icmptype=8),
qubes.firewall.Rule(action='accept', specialtarget='dns'),
]
self.vm.firewall.save()
expected_output = '\n'.join(rule.rule for rule in
self.vm.firewall.rules) + '\n'
with qubes.tests.tools.StdoutBuffer() as stdout:
qubes.tools.qvm_firewall.main(['--raw', self.vm.name, 'list'])
self.assertEqual(stdout.getvalue(), expected_output)
def test_010_add(self):
qubes.tools.qvm_firewall.main(
[self.vm.name, 'add', 'accept', '1.2.3.0/24', 'tcp', '443'])
self.assertEqual(self.vm.firewall.rules,
[qubes.firewall.Rule(action='accept', dsthost='1.2.3.0/24',
proto='tcp', dstports='443')])
def test_011_add_before(self):
self.vm.firewall.rules = [
qubes.firewall.Rule(action='accept', dsthost='1.2.3.1'),
qubes.firewall.Rule(action='accept', dsthost='1.2.3.2'),
qubes.firewall.Rule(action='accept', dsthost='1.2.3.3'),
]
self.vm.firewall.save()
qubes.tools.qvm_firewall.main(
[self.vm.name, 'add', '--before', '2',
'accept', '1.2.3.0/24', 'tcp', '443'])
self.vm.firewall.load()
self.assertEqual(self.vm.firewall.rules,
[qubes.firewall.Rule(action='accept', dsthost='1.2.3.1'),
qubes.firewall.Rule(action='accept', dsthost='1.2.3.2'),
qubes.firewall.Rule(action='accept', dsthost='1.2.3.0/24',
proto='tcp', dstports='443'),
qubes.firewall.Rule(action='accept', dsthost='1.2.3.3'),
])
def test_020_del(self):
self.vm.firewall.rules = [
qubes.firewall.Rule(action='accept', dsthost='1.2.3.1'),
qubes.firewall.Rule(action='accept', dsthost='1.2.3.2'),
qubes.firewall.Rule(action='accept', dsthost='1.2.3.3'),
]
self.vm.firewall.save()
qubes.tools.qvm_firewall.main(
[self.vm.name, 'del', 'accept', '1.2.3.2'])
self.vm.firewall.load()
self.assertEqual(self.vm.firewall.rules,
[qubes.firewall.Rule(action='accept', dsthost='1.2.3.1'),
qubes.firewall.Rule(action='accept', dsthost='1.2.3.3'),
])
def test_021_del_by_number(self):
self.vm.firewall.rules = [
qubes.firewall.Rule(action='accept', dsthost='1.2.3.1'),
qubes.firewall.Rule(action='accept', dsthost='1.2.3.2'),
qubes.firewall.Rule(action='accept', dsthost='1.2.3.3'),
]
self.vm.firewall.save()
qubes.tools.qvm_firewall.main(
[self.vm.name, 'del', '--rule-no', '1'])
self.vm.firewall.load()
self.assertEqual(self.vm.firewall.rules,
[qubes.firewall.Rule(action='accept', dsthost='1.2.3.1'),
qubes.firewall.Rule(action='accept', dsthost='1.2.3.3'),
])
def test_030_policy(self):
with qubes.tests.tools.StdoutBuffer() as stdout:
qubes.tools.qvm_firewall.main([self.vm.name, 'policy'])
self.assertEqual(stdout.getvalue(), 'accept\n')
self.vm.firewall.policy = 'drop'
self.vm.firewall.save()
with qubes.tests.tools.StdoutBuffer() as stdout:
qubes.tools.qvm_firewall.main([self.vm.name, 'policy'])
self.assertEqual(stdout.getvalue(), 'drop\n')
def test_031_policy_set(self):
qubes.tools.qvm_firewall.main([self.vm.name, 'policy', 'drop'])
self.assertEqual(self.vm.firewall.policy, 'drop')
qubes.tools.qvm_firewall.main([self.vm.name, 'policy', 'accept'])
self.vm.firewall.load()
self.assertEqual(self.vm.firewall.policy, 'accept')

View File

@ -1,57 +0,0 @@
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2015 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import qubes
import qubes.tools.qvm_prefs
import qubes.tests
@qubes.tests.skipUnlessDom0
class TC_00_qvm_prefs(
qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
def test_000_list(self):
self.assertEqual(0, qubes.tools.qvm_prefs.main([
'--qubesxml', qubes.tests.XMLPATH, 'dom0']))
def test_001_no_vm(self):
with self.assertRaises(SystemExit):
qubes.tools.qvm_prefs.main([
'--qubesxml', qubes.tests.XMLPATH])
def test_002_set_property(self):
self.assertEqual(0, qubes.tools.qvm_prefs.main([
'--qubesxml', qubes.tests.XMLPATH, 'dom0',
'default_user', 'testuser']))
self.assertEqual('testuser',
qubes.Qubes(qubes.tests.XMLPATH).domains['dom0'].default_user)
def test_003_invalid_property(self):
with self.assertRaises(SystemExit):
qubes.tools.qvm_prefs.main([
'--qubesxml', qubes.tests.XMLPATH, 'dom0',
'no_such_property'])
def test_004_set_invalid_property(self):
with self.assertRaises(SystemExit):
qubes.tools.qvm_prefs.main([
'--qubesxml', qubes.tests.XMLPATH, 'dom0',
'no_such_property', 'value'])

View File

@ -1,133 +0,0 @@
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2015 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import io
import os
import sys
import tempfile
import unittest
import qubes
import qubes.config
import qubes.tools.qvm_run
import qubes.vm
import qubes.tests
@qubes.tests.skipUnlessDom0
class TC_00_qvm_run(qubes.tests.SystemTestsMixin, qubes.tests.QubesTestCase):
def setUp(self):
super(TC_00_qvm_run, self).setUp()
self.init_default_template()
self.vm1 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('vm1'),
template=self.app.default_template,
label='red')
self.vm1.create_on_disk()
self.vm1.start()
self.app.save()
self.sharedopts = ['--qubesxml', qubes.tests.XMLPATH]
def tearDown(self):
# clean up after testing --colour-output
sys.stdout = sys.__stdout__
@staticmethod
def get_qvm_run_output(args):
assert '--localcmd' not in args, \
'get_qvm_run_output requires no --localcmd'
outfile = tempfile.NamedTemporaryFile(prefix='qvm-run-output')
args = list(args)
args.insert(0, '--pass-io')
args.insert(1, '--localcmd')
args.insert(2, 'sh -c "dd of={}"'.format(outfile.name))
qubes.tools.qvm_run.main(args)
outfile.seek(0)
output = outfile.read()
outfile.close()
return output
def test_000_basic(self):
self.assertEqual(0, qubes.tools.qvm_run.main(
self.sharedopts + [self.vm1.name, 'true']))
def test_001_passio_retcode(self):
self.assertEqual(0, qubes.tools.qvm_run.main(
self.sharedopts + ['--pass-io', self.vm1.name, 'true']))
self.assertEqual(1, qubes.tools.qvm_run.main(
self.sharedopts + ['--pass-io', self.vm1.name, 'false']))
def test_002_passio_localcmd(self):
self.assertEqual(b'aqq', self.get_qvm_run_output(
self.sharedopts + [self.vm1.name, 'printf aqq']))
def test_003_user(self):
self.assertNotEqual(b'0\n', self.get_qvm_run_output(
self.sharedopts + ['--user', 'user', self.vm1.name, 'id -u']))
self.assertEqual(b'0\n', self.get_qvm_run_output(
self.sharedopts + ['--user', 'root', self.vm1.name, 'id -u']))
def test_004_autostart(self):
vm2 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('vm2'),
template=qubes.tests.TEMPLATE,
label='red')
vm2.create_on_disk()
self.app.save()
# and do not start it
self.assertEqual(-1, qubes.tools.qvm_run.main(
self.sharedopts + [vm2.name, 'true']))
self.assertEqual(0, qubes.tools.qvm_run.main(
self.sharedopts + ['--autostart', vm2.name, 'true']))
@unittest.skip('expected error')
def test_005_colour_output(self):
sys.stdout = io.StringIO()
qubes.tools.qvm_run.main(
self.sharedopts + ['--colour-output', '32', self.vm1.name, 'true'])
self.assertEqual(b'\033[0;32m\033[0m', sys.stdout.getvalue())
def test_006_filter_esc(self):
self.assertEqual(b'\033', self.get_qvm_run_output(
self.sharedopts + ['--no-filter-escape-chars', self.vm1.name,
r'printf \\033']))
self.assertEqual(b'_', self.get_qvm_run_output(
self.sharedopts + ['--filter-escape-chars', self.vm1.name,
r'printf \\033']))
def test_007_gui(self): # pylint: disable=no-self-use
raise unittest.SkipTest('test not implemented')
#parser.add_argument('--gui',
#parser.add_argument('--no-gui', '--nogui',

View File

@ -1,140 +0,0 @@
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2015 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import argparse
import qubes
import qubes.tools
import qubes.tests
class TC_00_PropertyAction(qubes.tests.QubesTestCase):
def test_000_default(self):
parser = argparse.ArgumentParser()
parser.add_argument('--property', '-p',
action=qubes.tools.PropertyAction)
parser.set_defaults(properties={'defaultprop': 'defaultvalue'})
args = parser.parse_args([])
self.assertDictContainsSubset(
{'defaultprop': 'defaultvalue'}, args.properties)
def test_001_set_prop(self):
parser = argparse.ArgumentParser()
parser.add_argument('--property', '-p',
action=qubes.tools.PropertyAction)
args = parser.parse_args(['-p', 'testprop=testvalue'])
self.assertDictContainsSubset(
{'testprop': 'testvalue'}, args.properties)
def test_002_set_prop_2(self):
parser = argparse.ArgumentParser()
parser.add_argument('--property', '-p',
action=qubes.tools.PropertyAction)
parser.set_defaults(properties={'defaultprop': 'defaultvalue'})
args = parser.parse_args(
['-p', 'testprop=testvalue', '-p', 'testprop2=testvalue2'])
self.assertDictContainsSubset(
{'testprop': 'testvalue', 'testprop2': 'testvalue2'},
args.properties)
def test_003_set_prop_with_default(self):
parser = argparse.ArgumentParser()
parser.add_argument('--property', '-p',
action=qubes.tools.PropertyAction)
parser.set_defaults(properties={'defaultprop': 'defaultvalue'})
args = parser.parse_args(['-p', 'testprop=testvalue'])
self.assertDictContainsSubset(
{'testprop': 'testvalue', 'defaultprop': 'defaultvalue'},
args.properties)
def test_003_set_prop_override_default(self):
# pylint: disable=invalid-name
parser = argparse.ArgumentParser()
parser.add_argument('--property', '-p',
action=qubes.tools.PropertyAction)
parser.set_defaults(properties={'testprop': 'defaultvalue'})
args = parser.parse_args(['-p', 'testprop=testvalue'])
self.assertDictContainsSubset(
{'testprop': 'testvalue'},
args.properties)
class TC_01_SinglePropertyAction(qubes.tests.QubesTestCase):
def test_000_help(self):
parser = argparse.ArgumentParser()
action = parser.add_argument('--testprop', '-T',
action=qubes.tools.SinglePropertyAction)
self.assertIn('testprop', action.help)
def test_001_help_const(self):
parser = argparse.ArgumentParser()
action = parser.add_argument('--testprop', '-T',
action=qubes.tools.SinglePropertyAction,
const='testvalue')
self.assertIn('testvalue', action.help)
def test_100_default(self):
parser = argparse.ArgumentParser()
parser.add_argument('--testprop', '-T',
action=qubes.tools.SinglePropertyAction)
parser.set_defaults(properties={'testprop': 'defaultvalue'})
args = parser.parse_args([])
self.assertDictContainsSubset(
{'testprop': 'defaultvalue'}, args.properties)
def test_101_set_prop(self):
parser = argparse.ArgumentParser()
parser.add_argument('--testprop', '-T',
action=qubes.tools.SinglePropertyAction)
args = parser.parse_args(['-T', 'testvalue'])
self.assertDictContainsSubset(
{'testprop': 'testvalue'}, args.properties)
def test_102_set_prop_dest(self):
parser = argparse.ArgumentParser()
parser.add_argument('--testprop', '-T', dest='otherprop',
action=qubes.tools.SinglePropertyAction)
args = parser.parse_args(['-T', 'testvalue'])
self.assertDictContainsSubset(
{'otherprop': 'testvalue'}, args.properties)
def test_103_set_prop_const(self):
parser = argparse.ArgumentParser()
parser.add_argument('--testprop', '-T',
action=qubes.tools.SinglePropertyAction,
const='testvalue')
args = parser.parse_args(['-T'])
self.assertDictContainsSubset(
{'testprop': 'testvalue'}, args.properties)
def test_104_set_prop_positional(self):
parser = argparse.ArgumentParser()
parser.add_argument('testprop',
action=qubes.tools.SinglePropertyAction)
args = parser.parse_args(['testvalue'])
self.assertDictContainsSubset(
{'testprop': 'testvalue'}, args.properties)

View File

@ -1,158 +0,0 @@
# pylint: disable=protected-access
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2015 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
''' Tests for the `qvm-device` tool. '''
import qubes
import qubes.devices
import qubes.tools.qvm_device
import qubes.tests
import qubes.tests.devices
import qubes.tests.tools
class TestNamespace(object):
''' A mock object for `argparse.Namespace`.
''' # pylint: disable=too-few-public-methods
def __init__(self, app, domains=None, device=None):
super(TestNamespace, self).__init__()
self.app = app
self.devclass = 'testclass'
self.persistent = True
if domains:
self.domains = domains
if device:
self.device = device
self.device_assignment = qubes.devices.DeviceAssignment(
backend_domain=self.device.backend_domain,
ident=self.device.ident, persistent=self.persistent)
class TC_00_Actions(qubes.tests.QubesTestCase):
''' Tests the output logic of the qvm-device tool '''
def setUp(self):
super(TC_00_Actions, self).setUp()
self.app = qubes.tests.devices.TestApp()
def save():
''' A mock method for simulating a successful save '''
return True
self.app.save = save
self.vm1 = qubes.tests.devices.TestVM(self.app, 'vm1')
self.vm2 = qubes.tests.devices.TestVM(self.app, 'vm2')
self.device = self.vm2.device
def test_000_list_all(self):
''' List all exposed vm devices. No devices are attached to other
domains.
'''
args = TestNamespace(self.app)
with qubes.tests.tools.StdoutBuffer() as buf:
qubes.tools.qvm_device.list_devices(args)
self.assertEqual(
[x.rstrip() for x in buf.getvalue().splitlines()],
['vm1:testdev Description',
'vm2:testdev Description']
)
def test_001_list_persistent_attach(self):
''' Attach the device exposed by the `vm2` to the `vm1` persistently.
'''
args = TestNamespace(self.app, [self.vm1])
# simulate attach
assignment = qubes.devices.DeviceAssignment(backend_domain=self.vm2,
ident=self.device.ident, persistent=True, frontend_domain=self.vm1)
self.vm2.device.frontend_domain = self.vm1
self.vm1.devices['testclass']._set.add(assignment)
with qubes.tests.tools.StdoutBuffer() as buf:
qubes.tools.qvm_device.list_devices(args)
self.assertEqual(
buf.getvalue(),
'vm1:testdev Description\n'
'vm2:testdev Description vm1 vm1\n'
)
def test_002_list_list_temp_attach(self):
''' Attach the device exposed by the `vm2` to the `vm1`
non-persistently.
'''
args = TestNamespace(self.app, [self.vm1])
# simulate attach
assignment = qubes.devices.DeviceAssignment(backend_domain=self.vm2,
ident=self.device.ident, persistent=True, frontend_domain=self.vm1)
self.vm2.device.frontend_domain = self.vm1
self.vm1.devices['testclass']._set.add(assignment)
with qubes.tests.tools.StdoutBuffer() as buf:
qubes.tools.qvm_device.list_devices(args)
self.assertEqual(buf.getvalue(),
'vm1:testdev Description\n'
'vm2:testdev Description vm1 vm1\n')
def test_010_attach(self):
''' Test attach action '''
args = TestNamespace(
self.app,
[self.vm1],
self.device
)
qubes.tools.qvm_device.attach_device(args)
self.assertEventFired(self.vm1,
'device-attach:testclass', kwargs={'device': self.device})
self.assertEventNotFired(self.vm2,
'device-attach:testclass', kwargs={'device': self.device})
def test_011_double_attach(self):
''' Double attach should not be possible '''
args = TestNamespace(
self.app,
[self.vm1],
self.device
)
qubes.tools.qvm_device.attach_device(args)
with self.assertRaises(qubes.exc.QubesException):
qubes.tools.qvm_device.attach_device(args)
def test_020_detach(self):
''' Test detach action '''
args = TestNamespace(
self.app,
[self.vm1],
self.device
)
# simulate attach
self.vm2.device.frontend_domain = self.vm1
args.device_assignment.frontend_domain = self.vm1
self.vm1.devices['testclass']._set.add(args.device_assignment)
qubes.tools.qvm_device.detach_device(args)
def test_021_detach_not_attached(self):
''' Invalid detach action should not be possible '''
args = TestNamespace(
self.app,
[self.vm1],
self.device
)
with self.assertRaises(qubes.exc.QubesException):
qubes.tools.qvm_device.detach_device(args)

View File

@ -1,61 +0,0 @@
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2016 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
import argparse
import qubes.firewall
import qubes.tests
import qubes.tests.firewall
import qubes.tools.qvm_firewall
class TC_00_RuleAction(qubes.tests.QubesTestCase):
def setUp(self):
super(TC_00_RuleAction, self).setUp()
self.action = qubes.tools.qvm_firewall.RuleAction(None, dest='rule')
def test_000_named_opts(self):
ns = argparse.Namespace()
self.action(None, ns, ['dsthost=127.0.0.1', 'action=accept'])
self.assertEqual(ns.rule,
qubes.firewall.Rule(None, action='accept', dsthost='127.0.0.1/32'))
def test_001_unnamed_opts(self):
ns = argparse.Namespace()
self.action(None, ns, ['accept', '127.0.0.1', 'tcp', '80'])
self.assertEqual(ns.rule,
qubes.firewall.Rule(None, action='accept', dsthost='127.0.0.1/32',
proto='tcp', dstports=80))
def test_002_unnamed_opts(self):
ns = argparse.Namespace()
self.action(None, ns, ['accept', '127.0.0.1', 'icmp', '8'])
self.assertEqual(ns.rule,
qubes.firewall.Rule(None, action='accept', dsthost='127.0.0.1/32',
proto='icmp', icmptype=8))
def test_003_mixed_opts(self):
ns = argparse.Namespace()
self.action(None, ns, ['dsthost=127.0.0.1', 'accept',
'dstports=443', 'tcp'])
self.assertEqual(ns.rule,
qubes.firewall.Rule(None, action='accept', dsthost='127.0.0.1/32',
proto='tcp', dstports=443))

View File

@ -21,8 +21,6 @@
'''qvm-create - Create new Qubes OS store'''
# TODO allow to set properties and create domains
import sys
import qubes
import qubes.tools
@ -32,11 +30,6 @@ parser = qubes.tools.QubesArgumentParser(
want_app=True,
want_app_no_instance=True)
parser.add_argument('--property', '--prop', '-p',
action=qubes.tools.PropertyAction,
help='set global property')
def main(args=None):
'''Main routine of :program:`qubes-create`.
@ -46,7 +39,7 @@ def main(args=None):
args = parser.parse_args(args)
qubes.Qubes.create_empty_store(args.app,
offline_mode=args.offline_mode, **args.properties)
offline_mode=args.offline_mode)
return 0

View File

@ -1,113 +0,0 @@
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2010-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2015 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# TODO merge printing with qvm-prefs
# TODO list only non-default properties
from __future__ import print_function
import argparse
import sys
import qubes
import qubes.tools
import qubes.utils
parser = qubes.tools.QubesArgumentParser()
# keep it here for compatibility with earlier and possibly future versions
parser.add_argument('--force-root',
action='store_true', help=argparse.SUPPRESS)
parser.add_argument('--help-properties',
action=qubes.tools.HelpPropertiesAction)
parser.add_argument('--get', '-g',
action='store_true',
help='Ignored; for compatibility with older scripts.')
parser.add_argument('--set', '-s',
action='store_true',
help='Ignored; for compatibility with older scripts.')
parser.add_argument('property', metavar='PROPERTY',
nargs='?',
help='name of the property to show or change')
parser_value = parser.add_mutually_exclusive_group()
parser_value.add_argument('value', metavar='VALUE',
nargs='?',
help='new value of the property')
parser.add_argument('--unset', '--default', '--delete', '-D',
dest='delete',
action='store_true',
help='unset the property; if property has default value, it will be used'
' instead')
def main(args=None):
args = parser.parse_args(args)
if args.property is None:
properties = args.app.property_list()
width = max(len(prop.__name__) for prop in properties)
for prop in sorted(properties):
try:
value = getattr(args.app, prop.__name__)
except AttributeError:
print('{name:{width}s} U'.format(
name=prop.__name__, width=width))
continue
if args.app.property_is_default(prop):
print('{name:{width}s} D {value!s}'.format(
name=prop.__name__, width=width, value=value))
else:
print('{name:{width}s} - {value!s}'.format(
name=prop.__name__, width=width, value=value))
return 0
else:
args.property = args.property.replace('-', '_')
if args.value is not None:
setattr(args.app, args.property, args.value)
args.app.save()
return 0
if args.delete:
delattr(args.app, args.property)
args.app.save()
return 0
print(str(getattr(args.app, args.property)))
return 0
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,273 +0,0 @@
# pylint: disable=C,R
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2016 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
#
# 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
'''Qubes volume and block device managment'''
from __future__ import print_function
import sys
import qubes
import qubes.exc
import qubes.tools
import qubes.utils
def prepare_table(vd_list, full=False):
''' Converts a list of :py:class:`VolumeData` objects to a list of tupples
for the :py:func:`qubes.tools.print_table`.
If :program:`qvm-block` is running in a TTY, it will ommit duplicate
data.
:param list vd_list: List of :py:class:`VolumeData` objects.
:param bool full: If set to true duplicate data is printed even when
running from TTY.
:returns: list of tupples
'''
output = []
if sys.stdout.isatty():
# NOQA
output += [('POOL:VOLUME', 'VMNAME', 'VOLUME_NAME', 'REVERT_POSSIBLE')]
for volume in vd_list:
if volume.domains:
vmname, volume_name = volume.domains.pop()
output += [(str(volume), vmname, volume_name, volume.revisions)]
for tupple in volume.domains:
vmname, volume_name = tupple
if full or not sys.stdout.isatty():
output += [(str(volume), vmname, volume_name,
volume.revisions)]
else:
output += [('', vmname, volume_name, '', volume.revisions)]
else:
output += [(str(volume), "")]
return output
class VolumeData(object):
''' Wrapper object around :py:class:`qubes.storage.Volume`, mainly to track
the domains a volume is attached to.
'''
# pylint: disable=too-few-public-methods
def __init__(self, volume):
self.name = volume.name
self.pool = volume.pool
self.vid = volume.vid
if volume.revisions != {}:
self.revisions = 'Yes'
else:
self.revisions = 'No'
self.domains = []
def __str__(self):
return "{!s}:{!s}".format(self.pool, self.vid)
def list_volumes(args):
''' Called by the parser to execute the qubes-block list subcommand. '''
app = args.app
if args.pools:
pools = args.pools # only specified pools
else:
pools = app.pools.values() # all pools
volumes = [v for p in pools for v in p.volumes]
if not args.internal: # hide internal volumes
volumes = [v for v in volumes if not v.internal]
vd_dict = {}
for volume in volumes:
volume_data = VolumeData(volume)
try:
vd_dict[volume.pool][volume.vid] = volume_data
except KeyError:
vd_dict[volume.pool] = {volume.vid: volume_data}
if hasattr(args, 'domains') and args.domains:
domains = args.domains
else:
domains = args.app.domains
for domain in domains: # gather the domain names
try:
for volume in domain.attached_volumes:
try:
if not args.internal and volume.internal:
# some pools (LVM) may set 'internal' flag only when
# listing volumes of specific domain
del vd_dict[volume.pool][volume.vid]
else:
volume_data = vd_dict[volume.pool][volume.vid]
volume_data.domains += [(domain.name, volume.name)]
except KeyError:
# Skipping volume
continue
except AttributeError:
# Skipping domain without volumes
continue
if hasattr(args, 'domains') and args.domains:
result = [x # reduce to only VolumeData with assigned domains
for p in vd_dict.values() for x in p.values()
if x.domains]
else:
result = [x for p in vd_dict.values() for x in p.values()]
qubes.tools.print_table(prepare_table(result, full=args.full))
def revert_volume(args):
volume = args.volume
app = args.app
try:
pool = app.pools[volume.pool]
pool.revert(volume)
except qubes.storage.StoragePoolException as e:
print(str(e), file=sys.stderr)
sys.exit(1)
def attach_volumes(args):
''' Called by the parser to execute the :program:`qvm-block attach`
subcommand.
'''
volume = args.volume
vm = args.domains[0]
try:
rw = not args.ro
vm.storage.attach(volume, rw=rw)
except qubes.storage.StoragePoolException as e:
print(str(e), file=sys.stderr)
sys.exit(1)
def detach_volumes(args):
''' Called by the parser to execute the :program:`qvm-block detach`
subcommand.
'''
volume = args.volume
vm = args.domains[0]
try:
vm.storage.detach(volume)
except qubes.storage.StoragePoolException as e:
print(str(e), file=sys.stderr)
sys.exit(1)
def extend_volumes(args):
''' Called by the parser to execute the :program:`qvm-block extend`
subcommand
'''
volume = args.volume
app = args.app
size = qubes.utils.parse_size(args.size)
pool = app.get_pool(volume.pool)
pool.resize(volume, volume.size+size)
app.save()
def init_list_parser(sub_parsers):
''' Configures the parser for the :program:`qvm-block list` subcommand '''
# pylint: disable=protected-access
list_parser = sub_parsers.add_parser('list', aliases=('ls', 'l'),
help='list block devices')
list_parser.add_argument('-p', '--pool', dest='pools',
action=qubes.tools.PoolsAction)
list_parser.add_argument('-i', '--internal', action='store_true',
help='Show internal volumes')
list_parser.add_argument(
'--full', action='store_true',
help='print full line for each POOL_NAME:VOLUME_ID & vm combination')
vm_name_group = qubes.tools.VmNameGroup(
list_parser, required=False, vm_action=qubes.tools.VmNameAction,
help='list volumes from specified domain(s)')
list_parser._mutually_exclusive_groups.append(vm_name_group)
list_parser.set_defaults(func=list_volumes)
def init_revert_parser(sub_parsers):
revert_parser = sub_parsers.add_parser(
'revert', aliases=('rv', 'r'),
help='revert volume to previous revision')
revert_parser.add_argument(metavar='POOL_NAME:VOLUME_ID', dest='volume',
action=qubes.tools.VolumeAction)
revert_parser.set_defaults(func=revert_volume)
def init_attach_parser(sub_parsers):
attach_parser = sub_parsers.add_parser(
'attach', help="Attach volume to domain", aliases=('at', 'a'))
attach_parser.add_argument('--ro', help='attach device read-only',
action='store_true')
attach_parser.add_argument('VMNAME', action=qubes.tools.RunningVmNameAction)
attach_parser.add_argument(metavar='POOL_NAME:VOLUME_ID', dest='volume',
action=qubes.tools.VolumeAction)
attach_parser.set_defaults(func=attach_volumes)
def init_dettach_parser(sub_parsers):
detach_parser = sub_parsers.add_parser(
"detach", help="Detach volume from domain", aliases=('d', 'dt'))
detach_parser.add_argument('VMNAME', action=qubes.tools.RunningVmNameAction)
detach_parser.add_argument(metavar='POOL_NAME:VOLUME_ID', dest='volume',
action=qubes.tools.VolumeAction)
detach_parser.set_defaults(func=detach_volumes)
def init_extend_parser(sub_parsers):
extend_parser = sub_parsers.add_parser(
"extend", help="extend volume from domain", aliases=('d', 'dt'))
extend_parser.add_argument(metavar='POOL_NAME:VOLUME_ID', dest='volume',
action=qubes.tools.VolumeAction)
extend_parser.add_argument('size', help='New size in bytes')
extend_parser.set_defaults(func=extend_volumes)
def get_parser():
'''Create :py:class:`argparse.ArgumentParser` suitable for
:program:`qvm-block`.
'''
parser = qubes.tools.QubesArgumentParser(description=__doc__, want_app=True)
parser.register('action', 'parsers', qubes.tools.AliasedSubParsersAction)
sub_parsers = parser.add_subparsers(
title='commands',
description="For more information see qvm-block command -h",
dest='command')
init_attach_parser(sub_parsers)
init_dettach_parser(sub_parsers)
init_extend_parser(sub_parsers)
init_list_parser(sub_parsers)
init_revert_parser(sub_parsers)
return parser
def main(args=None):
'''Main routine of :program:`qvm-block`.'''
parser = get_parser()
try:
args = parser.parse_args(args)
args.func(args)
except qubes.exc.QubesException as e:
parser.print_error(str(e))
return 1
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,75 +0,0 @@
# pylint: disable=too-few-public-methods
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2016 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
#
# 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
''' Exits sucessfull if the provided domains exists, else returns failure '''
from __future__ import print_function
import sys
import qubes.tools
import qubes.vm.templatevm
parser = qubes.tools.QubesArgumentParser(description=__doc__, vmname_nargs='+')
parser.add_argument("--running", action="store_true", dest="running",
default=False, help="Determine if (any of given) VM is running")
parser.add_argument("--paused", action="store_true", dest="paused",
default=False, help="Determine if (any of given) VM is paused")
parser.add_argument("--template", action="store_true", dest="template",
default=False, help="Determine if (any of given) VM is a template")
def print_msg(domains, what_single, what_plural):
if not domains:
print("None of given VM {!s}".format(what_single))
elif len(domains) == 1:
print("VM {!s} {!s}".format(domains[0], what_single))
else:
txt = ", ".join([vm.name for vm in domains])
print("VMs {!s} {!s}".format(txt, what_plural))
def main(args=None):
args = parser.parse_args(args)
domains = args.domains
if args.running:
running = [vm for vm in domains if vm.is_running()]
if args.verbose:
print_msg(running, "is running", "are running")
return 0 if running else 1
elif args.paused:
paused = [vm for vm in domains if vm.is_paused()]
if args.verbose:
print_msg(paused, "is paused", "are paused")
return 0 if paused else 1
elif args.template:
template = [vm for vm in domains if isinstance(vm,
qubes.vm.templatevm.TemplateVM)]
if args.verbose:
print_msg(template, "is a template", "are templates")
return 0 if template else 1
else:
if args.verbose:
print_msg(domains, "exists", "exist")
return 0 if domains else 1
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,71 +0,0 @@
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2016 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
#
# 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
''' Clone a domain '''
import sys
from qubes.tools import QubesArgumentParser, SinglePropertyAction
parser = QubesArgumentParser(description=__doc__, vmname_nargs=1)
parser.add_argument('new_name',
metavar='NEWVM',
action=SinglePropertyAction,
help='name of the domain to create')
group = parser.add_mutually_exclusive_group()
group.add_argument('-P',
metavar='POOL',
dest='one_pool',
default='',
help='pool to use for the new domain')
group.add_argument('-p',
'--pool',
action='append',
metavar='POOL:VOLUME',
help='specify the pool to use for the specific volume')
def main(args=None):
''' Clones an existing VM by copying all its disk files '''
args = parser.parse_args(args)
app = args.app
src_vm = args.domains[0]
new_name = args.properties['new_name']
dst_vm = app.add_new_vm(src_vm.__class__, name=new_name)
dst_vm.clone_properties(src_vm)
if args.one_pool:
dst_vm.clone_disk_files(src_vm, pool=args.one_pool)
elif hasattr(args, 'pools') and args.pools:
dst_vm.clone_disk_files(src_vm, pools=args.pools)
else:
dst_vm.clone_disk_files(src_vm)
# try:
app.save() # HACK remove_from_disk on exception hangs for some reason
# except Exception as e: # pylint: disable=broad-except
# dst_vm.remove_from_disk()
# parser.print_error(e)
# return 0
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,179 +0,0 @@
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2010-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2015 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# TODO list available classes
# TODO list labels (maybe in qvm-prefs)
# TODO features, devices, tags
from __future__ import print_function
import argparse
import os
import subprocess
import sys
import qubes
import qubes.tools
parser = qubes.tools.QubesArgumentParser(want_force_root=True)
parser.add_argument('--class', '-C', dest='cls',
default='AppVM',
help='specify the class of the new domain (default: %(default)s)')
parser.add_argument('--property', '--prop',
action=qubes.tools.PropertyAction,
help='set domain\'s property, like "internal", "memory" or "vcpus"')
parser.add_argument('--pool', '-p',
action='append',
metavar='POOL_NAME:VOLUME_NAME',
help='specify the pool to use for a volume')
parser.add_argument('-P',
metavar='POOL_NAME',
dest='one_pool',
default='',
help='change all volume pools to specified pool')
parser.add_argument('--template', '-t',
action=qubes.tools.SinglePropertyAction,
help='specify the TemplateVM to use')
parser.add_argument('--label', '-l',
action=qubes.tools.SinglePropertyAction,
help='specify the label to use for the new domain'
' (e.g. red, yellow, green, ...)')
parser_root = parser.add_mutually_exclusive_group()
parser_root.add_argument('--root-copy-from', '-r', metavar='FILENAME',
help='use provided root.img instead of default/empty one'
' (file will be COPIED)')
parser_root.add_argument('--root-move-from', '-R', metavar='FILENAME',
help='use provided root.img instead of default/empty one'
' (file will be MOVED)')
parser_root.add_argument('--no-root',
action='store_true', default=False,
help=argparse.SUPPRESS)
parser.add_argument('name', metavar='VMNAME',
action=qubes.tools.SinglePropertyAction,
nargs='?',
help='name of the domain to create')
def main(args=None):
args = parser.parse_args(args)
pools = {}
pool = None
if hasattr(args, 'pools') and args.pools:
for pool_vol in args.pool:
try:
pool_name, volume_name = pool_vol.split(':')
pools[volume_name] = pool_name
except ValueError:
parser.error(
'Pool argument must be of form: -P pool_name:volume_name')
if args.one_pool:
pool = args.one_pool
if 'label' not in args.properties:
parser.error('--label option is mandatory')
if 'name' not in args.properties:
parser.error('VMNAME is mandatory')
try:
args.app.get_label(args.properties['label'])
except KeyError:
parser.error('no such label: {!r}; available: {}'.format(
args.properties['label'],
', '.join(repr(l.name) for l in args.app.labels)))
try:
cls = args.app.get_vm_class(args.cls)
except KeyError:
parser.error('no such domain class: {!r}'.format(args.cls))
if 'template' in args.properties and \
'template' not in (prop.__name__ for prop in cls.property_list()):
parser.error('this domain class does not support template')
vm = args.app.add_new_vm(cls, **args.properties)
# pylint: disable=line-too-long
# if not options.standalone and any([options.root_copy_from, options.root_move_from]):
# print >> sys.stderr, "root.img can be specified only for standalone VMs"
# exit (1)
# if options.hvm_template and options.template is not None:
# print >> sys.stderr, "Template VM cannot be based on another template"
# exit (1)
# if options.root_copy_from is not None and not os.path.exists(options.root_copy_from):
# print >> sys.stderr, "File specified as root.img does not exists"
# exit (1)
# if options.root_move_from is not None and not os.path.exists(options.root_move_from):
# print >> sys.stderr, "File specified as root.img does not exists"
# exit (1)
# elif not options.hvm and not options.hvm_template:
# if qvm_collection.get_default_template() is None:
# print >> sys.stderr, "No default TemplateVM defined!"
# exit (1)
# else:
# template = qvm_collection.get_default_template()
# if (options.verbose):
# print('--> Using default TemplateVM: {0}'.format(template.name))
if not args.no_root:
try:
vm.create_on_disk(pool, pools)
# TODO this is file pool specific. Change it to a more general
# solution
root_img_path = vm.volumes['root'].vid
if args.root_move_from is not None:
# if (options.verbose):
# print "--> Replacing root.img with provided file"
os.unlink(root_img_path)
os.rename(args.root_move_from, root_img_path)
elif args.root_copy_from is not None:
# if (options.verbose):
# print "--> Replacing root.img with provided file"
os.unlink(root_img_path)
# use 'cp' to preserve sparse file
subprocess.check_call(['cp', args.root_copy_from, root_img_path])
except (IOError, OSError) as err:
parser.error(str(err))
args.app.save()
return 0
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,261 +0,0 @@
# pylint: disable=C,R
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2016 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
# Copyright (C) 2016 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
'''Qubes volume and block device managment'''
from __future__ import print_function
import argparse
import os
import sys
import qubes
import qubes.devices
import qubes.exc
import qubes.tools
def prepare_table(dev_list):
''' Converts a list of :py:class:`qubes.devices.DeviceInfo` objects to a
list of tupples for the :py:func:`qubes.tools.print_table`.
If :program:`qvm-devices` is running in a TTY, it will ommit duplicate
data.
:param list dev_list: List of :py:class:`qubes.devices.DeviceInfo`
objects.
:returns: list of tupples
'''
output = []
header = []
if sys.stdout.isatty():
header += [('VMNAME:DEVID', 'DESCRIPTION', 'USED BY', 'ASSIGNED')] # NOQA
for dev in dev_list:
output += [(
dev.id,
dev.description,
str(dev.attached_to),
dev.assignments
)]
return header + sorted(output)
class Line(object):
def __init__(self, device: qubes.devices.DeviceInfo, attached_to = None):
self.id = "{!s}:{!s}".format(device.backend_domain, device.ident)
self.description = device.description
self.attached_to = attached_to if attached_to else ""
self.frontends = []
@property
def assignments(self):
return ', '.join(self.frontends)
def list_devices(args):
''' Called by the parser to execute the qubes-devices list
subcommand. '''
app = args.app
result = []
devices = set()
if hasattr(args, 'domains') and args.domains:
for domain in args.domains:
for dev in domain.devices[args.devclass].attached():
devices.add(dev)
for dev in domain.devices[args.devclass].available():
devices.add(dev)
else:
for domain in app.domains:
for dev in domain.devices[args.devclass].available():
devices.add(dev)
result = {dev: Line(dev) for dev in devices}
for dev in result:
for domain in app.domains:
if domain == dev.backend_domain:
continue
elif dev in domain.devices[args.devclass].attached():
result[dev].attached_to = str(domain)
if dev in domain.devices[args.devclass].assignments():
if dev in domain.devices[args.devclass].persistent():
result[dev].frontends.append(str(domain))
qubes.tools.print_table(prepare_table(result.values()))
def attach_device(args):
''' Called by the parser to execute the :program:`qvm-devices attach`
subcommand.
'''
device_assignment = args.device_assignment
vm = args.domains[0]
app = args.app
device_assignment.persistent = args.persistent
vm.devices[args.devclass].attach(device_assignment)
if device_assignment.persistent:
app.save()
def detach_device(args):
''' Called by the parser to execute the :program:`qvm-devices detach`
subcommand.
'''
device_assignment = args.device_assignment
vm = args.domains[0]
before = len(vm.devices[args.devclass].persistent())
vm.devices[args.devclass].detach(device_assignment)
after = len(vm.devices[args.devclass].persistent())
if after < before:
args.app.save()
def init_list_parser(sub_parsers):
''' Configures the parser for the :program:`qvm-devices list` subcommand '''
# pylint: disable=protected-access
list_parser = sub_parsers.add_parser('list', aliases=('ls', 'l'),
help='list devices')
vm_name_group = qubes.tools.VmNameGroup(
list_parser, required=False, vm_action=qubes.tools.VmNameAction,
help='list devices assigned to specific domain(s)')
list_parser._mutually_exclusive_groups.append(vm_name_group)
list_parser.set_defaults(func=list_devices)
class DeviceAction(qubes.tools.QubesAction):
''' Action for argument parser that gets the
:py:class:``qubes.device.DeviceInfo`` from a BACKEND:DEVICE_ID string.
''' # pylint: disable=too-few-public-methods
def __init__(self, help='A pool & volume id combination',
required=True, **kwargs):
# pylint: disable=redefined-builtin
super(DeviceAction, self).__init__(help=help, required=required,
**kwargs)
def __call__(self, parser, namespace, values, option_string=None):
''' Set ``namespace.vmname`` to ``values`` '''
setattr(namespace, self.dest, values)
def parse_qubes_app(self, parser, namespace):
assert hasattr(namespace, 'app')
app = namespace.app
assert hasattr(namespace, 'device')
backend_device_id = getattr(namespace, self.dest)
assert hasattr(namespace, 'devclass')
devclass = namespace.devclass
try:
vmname, device_id = backend_device_id.split(':', 1)
try:
vm = app.domains[vmname]
except KeyError:
parser.error_runtime("no backend vm {!r}".format(vmname))
try:
vm.devices[devclass][device_id]
except KeyError:
parser.error_runtime(
"backend vm {!r} doesn't expose device {!r}"
.format(vmname, device_id))
device_assignment = qubes.devices.DeviceAssignment(vm, device_id,)
setattr(namespace, 'device_assignment', device_assignment)
except ValueError:
parser.error('expected a backend vm & device id combination ' \
'like foo:bar got %s' % backend_device_id)
def get_parser(device_class=None):
'''Create :py:class:`argparse.ArgumentParser` suitable for
:program:`qvm-block`.
'''
parser = qubes.tools.QubesArgumentParser(description=__doc__, want_app=True)
parser.register('action', 'parsers', qubes.tools.AliasedSubParsersAction)
if device_class:
parser.add_argument('devclass', const=device_class,
action='store_const',
help=argparse.SUPPRESS)
else:
parser.add_argument('devclass', metavar='DEVICE_CLASS', action='store',
help="Device class to manage ('pci', 'usb', etc)")
sub_parsers = parser.add_subparsers(
title='commands',
description="For more information see qvm-device command -h",
dest='command')
init_list_parser(sub_parsers)
attach_parser = sub_parsers.add_parser(
'attach', help="Attach device to domain", aliases=('at', 'a'))
detach_parser = sub_parsers.add_parser(
"detach", help="Detach device from domain", aliases=('d', 'dt'))
attach_parser.add_argument('VMNAME', action=qubes.tools.VmNameAction)
detach_parser.add_argument('VMNAME', action=qubes.tools.VmNameAction)
if device_class == 'block':
attach_parser.add_argument(metavar='BACKEND:DEVICE_ID', dest='device',
action=qubes.tools.VolumeAction)
detach_parser.add_argument(metavar='BACKEND:DEVICE_ID', dest='device',
action=qubes.tools.VolumeAction)
else:
attach_parser.add_argument(metavar='BACKEND:DEVICE_ID',
dest='device',
action=DeviceAction)
attach_parser.add_argument('-p', '--persistent', default=False,
help='device will attached on each start of the VMNAME',
action='store_true')
detach_parser.add_argument(metavar='BACKEND:DEVICE_ID',
dest='device',
action=DeviceAction)
attach_parser.set_defaults(func=attach_device)
detach_parser.set_defaults(func=detach_device)
return parser
def main(args=None):
'''Main routine of :program:`qvm-block`.'''
basename = os.path.basename(sys.argv[0])
devclass = None
if basename.startswith('qvm-') and basename != 'qvm-device':
devclass = basename[4:]
args = get_parser(devclass).parse_args(args)
try:
args.func(args)
except qubes.exc.QubesException as e:
print(str(e), file=sys.stderr)
return 1
return 0
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,102 +0,0 @@
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2010-2016 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2016 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
'''qvm-features - Manage domain's features'''
from __future__ import print_function
import argparse
import sys
import qubes
parser = qubes.tools.QubesArgumentParser(
vmname_nargs=1,
description='manage domain\'s features')
parser.add_argument('--request',
action='store_true', default=False,
help=argparse.SUPPRESS)
parser.add_argument('feature', metavar='FEATURE',
action='store', nargs='?',
help='name of the feature')
parser.add_argument('value', metavar='VALUE',
action='store', nargs='?',
help='new value of the feature')
parser.add_argument('--unset', '--default', '--delete', '-D',
dest='delete',
action='store_true',
help='unset the feature')
def main(args=None):
'''Main routine of :program:`qvm-features`.
:param list args: Optional arguments to override those delivered from \
command line.
'''
args = parser.parse_args(args)
vm = args.domains[0]
if args.request:
# Request mode: instead of setting the features directly,
# let the extensions handle them first.
vm.fire_event('feature-request', untrusted_features=args.features)
elif args.feature is None:
if args.delete:
parser.error('--unset requires a feature')
# max doesn't like empty list
if vm.features:
width = max(len(feature) for feature in vm.features)
for feature in sorted(vm.features):
print('{name:{width}s} {value}'.format(
name=feature, value=vm.features[feature], width=width))
elif args.delete:
if args.value is not None:
parser.error('cannot both set and unset a value')
try:
del vm.features[args.feature]
args.app.save()
except KeyError:
pass
elif args.value is None:
try:
print(vm.features[args.feature])
return 0
except KeyError:
return 1
else:
vm.features[args.feature] = args.value
args.app.save()
return 0
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,170 +0,0 @@
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2016 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
from __future__ import print_function
import argparse
import sys
import itertools
import qubes.firewall
import qubes.tools
class RuleAction(argparse.Action):
# pylint: disable=too-few-public-methods
'''Parser action for a single firewall rule. It accept syntax:
- <action> [<dsthost> [<proto> [<dstports>|<icmptype>]]]
- action=<action> [specialtarget=dns] [dsthost=<dsthost>]
[proto=<proto>] [dstports=<dstports>] [icmptype=<icmptype>]
Or a mix of them.
'''
def __call__(self, _parser, namespace, values, option_string=None):
if not values:
setattr(namespace, self.dest, None)
return
assumed_order = ['action', 'dsthost', 'proto', 'dstports', 'icmptype']
allowed_opts = assumed_order + ['specialtarget']
kwargs = {}
for opt in values:
opt_elements = opt.split('=')
if len(opt_elements) == 2:
key, value = opt_elements
elif len(opt_elements) == 1:
key, value = assumed_order[0], opt
else:
raise argparse.ArgumentError(None,
'invalid rule description: {}'.format(opt))
if key not in allowed_opts:
raise argparse.ArgumentError(None,
'Invalid rule element: {}'.format(opt))
kwargs[key] = value
if key in assumed_order:
assumed_order.remove(key)
if key == 'proto' and value in ['tcp', 'udp']:
assumed_order.remove('icmptype')
elif key == 'proto' and value in ['icmp']:
assumed_order.remove('dstports')
rule = qubes.firewall.Rule(**kwargs)
setattr(namespace, self.dest, rule)
parser = qubes.tools.QubesArgumentParser(vmname_nargs=1)
action = parser.add_subparsers(dest='command', help='action to perform')
action_add = action.add_parser('add', help='add rule')
action_add.add_argument('--before', type=int, default=None,
help='Add rule before rule with given number, instead of at the end')
action_add.add_argument('rule', nargs='+', action=RuleAction,
help='rule description')
action_del = action.add_parser('del', help='remove rule')
action_del.add_argument('--rule-no', dest='rule_no', type=int,
action='store', help='rule number')
action_del.add_argument('rule', nargs='*', action=RuleAction,
help='rule to be removed')
action_list = action.add_parser('list', help='list rules')
action_policy = action.add_parser('policy',
help='get/set policy - default action')
action_policy.add_argument('policy', choices=['accept', 'drop'],
help='policy value', default=None, nargs='?')
parser.add_argument('--reload', '-r', action='store_true',
help='force reloading rules even when unchanged')
parser.add_argument('--raw', action='store_true',
help='output rules as raw strings, instead of nice table')
def rules_list_table(vm):
header = ['NO', 'ACTION', 'HOST', 'PROTOCOL', 'PORT(S)',
'SPECIAL TARGET', 'ICMP TYPE']
rows = []
for (rule, rule_no) in zip(vm.firewall.rules, itertools.count()):
row = [str(x) if x is not None else '' for x in [
rule_no,
rule.action,
rule.dsthost,
rule.proto,
rule.dstports,
rule.specialtarget,
rule.icmptype,
]]
rows.append(row)
qubes.tools.print_table([header] + rows)
def rules_list_raw(vm):
for rule in vm.firewall.rules:
sys.stdout.write(rule.rule + '\n')
def rules_add(vm, args):
if args.before is not None:
vm.firewall.rules.insert(args.before, args.rule)
else:
vm.firewall.rules.append(args.rule)
vm.firewall.save()
def rules_del(vm, args):
if args.rule_no is not None:
vm.firewall.rules.pop(args.rule_no)
else:
vm.firewall.rules.remove(args.rule)
vm.firewall.save()
def policy(vm, args):
if args.policy is not None:
vm.firewall.policy = args.policy
vm.firewall.save()
else:
print(vm.firewall.policy)
def main(args=None):
'''Main routine of :program:`qvm-firewall`.'''
try:
args = parser.parse_args(args)
vm = args.domains[0]
if args.command == 'add':
rules_add(vm, args)
elif args.command == 'del':
rules_del(vm, args)
elif args.command == 'policy':
policy(vm, args)
elif args.command == 'list':
if args.raw:
rules_list_raw(vm)
else:
rules_list_table(vm)
if args.reload:
vm.fire_event('firewall-changed')
except qubes.exc.QubesException as e:
parser.print_error(str(e))
return 1
return 0
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,54 +0,0 @@
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2015 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
'''qvm-kill - forceful shutdown'''
import sys
import qubes.exc
import qubes.tools
parser = qubes.tools.QubesArgumentParser(
description='forceful shutdown of a domain', vmname_nargs='+')
def main(args=None):
'''Main routine of :program:`qvm-kill`.
:param list args: Optional arguments to override those delivered from \
command line.
'''
args = parser.parse_args(args)
exit_code = 0
for domain in args.domains:
try:
domain.force_shutdown()
except (IOError, OSError, qubes.exc.QubesException) as e:
exit_code = 1
parser.print_error(str(e))
return exit_code
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,47 +0,0 @@
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2010-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2015 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
'''qvm-pause - Pause a domain'''
import sys
import qubes
parser = qubes.tools.QubesArgumentParser(vmname_nargs='+',
description='pause a domain')
def main(args=None):
'''Main routine of :program:`qvm-pause`.
:param list args: Optional arguments to override those delivered from \
command line.
'''
args = parser.parse_args(args)
for domain in args.domains:
domain.pause()
return 0
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,214 +0,0 @@
# pylint: disable=too-few-public-methods
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2016 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
#
# 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
'''Manages Qubes pools and their options'''
from __future__ import print_function
import argparse
import sys
import qubes
import qubes.ext
import qubes.storage
import qubes.tools
drivers = qubes.storage.pool_drivers()
class _HelpDrivers(argparse.Action):
''' Action for argument parser that displays all drivers and their options
and exits.
'''
def __init__(self,
option_strings,
dest=argparse.SUPPRESS,
default=argparse.SUPPRESS):
super(_HelpDrivers, self).__init__(
option_strings=option_strings,
dest=dest,
default=default,
nargs=0,
help='list all drivers with their options and exit')
def __call__(self, parser, namespace, values, option_string=None):
result = []
for driver in drivers:
params = driver_parameters(driver)
driver_options = ', '.join(params)
result += [(driver, 'driver options', driver_options)]
qubes.tools.print_table(result)
parser.exit(0)
class _Info(qubes.tools.PoolsAction):
''' Action for argument parser that displays pool info and exits. '''
def __init__(self, option_strings, help='print pool info and exit',
**kwargs):
# pylint: disable=redefined-builtin
super(_Info, self).__init__(option_strings, help=help, **kwargs)
def __call__(self, parser, namespace, values, option_string=None):
setattr(namespace, 'command', 'info')
super(_Info, self).__call__(parser, namespace, values, option_string)
def pool_info(pool):
''' Prints out pool name and config '''
data = [("name", pool.name)]
data += [i for i in pool.config.items() if i[0] != 'name']
qubes.tools.print_table(data)
def list_pools(app):
''' Prints out all known pools and their drivers '''
result = [('NAME', 'DRIVER')]
for pool in app.pools.values():
if not pool.volumes and issubclass(
pool.__class__, qubes.storage.domain.DomainPool):
# skip empty DomainPools
continue
result += [(pool.name, pool.driver)]
qubes.tools.print_table(result)
class _Remove(argparse.Action):
''' Action for argument parser that removes a pool '''
def __init__(self, option_strings, dest=None, default=None, metavar=None):
super(_Remove, self).__init__(option_strings=option_strings,
dest=dest,
metavar=metavar,
default=default,
help='remove pool')
def __call__(self, parser, namespace, name, option_string=None):
setattr(namespace, 'command', 'remove')
setattr(namespace, 'name', name)
class _Add(argparse.Action):
''' Action for argument parser that adds a pool. '''
def __init__(self, option_strings, dest=None, default=None, metavar=None):
super(_Add, self).__init__(option_strings=option_strings,
dest=dest,
metavar=metavar,
default=default,
nargs=2,
help='add pool')
def __call__(self, parser, namespace, values, option_string=None):
name, driver = values
if driver not in drivers:
parser.error('driver %s is unknown \n' % driver)
else:
setattr(namespace, 'command', 'add')
setattr(namespace, 'name', name)
setattr(namespace, 'driver', driver)
class _Options(argparse.Action):
''' Action for argument parser that parsers options. '''
def __init__(self, option_strings, dest, default, metavar='options'):
super(_Options, self).__init__(
option_strings=option_strings,
dest=dest,
metavar=metavar,
default=default,
help='comma-separated list of driver options')
def __call__(self, parser, namespace, options, option_string=None):
setattr(namespace, 'options',
dict([option.split('=', 1) for option in options.split(',')]))
def get_parser():
''' Parses the provided args '''
epilog = 'available pool drivers: ' \
+ ', '.join(drivers)
parser = qubes.tools.QubesArgumentParser(description=__doc__,
epilog=epilog)
parser.add_argument('--help-drivers', action=_HelpDrivers)
parser.add_argument('-o', action=_Options, dest='options', default={})
group = parser.add_mutually_exclusive_group()
group.add_argument('-l',
'--list',
dest='command',
const='list',
action='store_const',
help='list all pools and exit (default action)')
group.add_argument('-i', '--info', metavar='POOLNAME', dest='pools',
action=_Info, default=[])
group.add_argument('-a',
'--add',
action=_Add,
dest='command',
metavar=('NAME', 'DRIVER'))
group.add_argument('-r', '--remove', metavar='NAME', action=_Remove)
return parser
def driver_parameters(name):
''' Get __init__ parameters from a driver with out `self` & `name`. '''
init_function = qubes.utils.get_entry_point_one(
qubes.storage.STORAGE_ENTRY_POINT, name).__init__
params = init_function.func_code.co_varnames
ignored_params = ['self', 'name']
return [p for p in params if p not in ignored_params]
def main(args=None):
'''Main routine of :program:`qvm-pools`.
:param list args: Optional arguments to override those delivered from \
command line.
'''
parser = get_parser()
try:
args = parser.parse_args(args)
except qubes.exc.QubesException as e:
parser.print_error(str(e))
return 1
if args.command is None or args.command == 'list':
list_pools(args.app)
elif args.command == 'add':
if args.name in args.app.pools.keys():
parser.error('pool named %s already exists \n' % args.name)
args.app.add_pool(name=args.name, driver=args.driver, **args.options)
args.app.save()
elif args.command == 'remove':
if args.name in args.app.pools.keys():
args.app.remove_pool(args.name)
args.app.save()
else:
parser.print_error('no such pool %s\n' % args.name)
elif args.command == 'info':
pool_info(args.pools)
return 0
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,115 +0,0 @@
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2010-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2015 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# TODO list properties for all classes
# TODO list only non-default properties
from __future__ import print_function
import sys
import qubes
import qubes.tools
import qubes.utils
import qubes.vm
parser = qubes.tools.QubesArgumentParser(
want_force_root=True,
vmname_nargs=1)
parser.add_argument('--help-properties',
action=qubes.tools.HelpPropertiesAction,
klass=qubes.vm.qubesvm.QubesVM)
parser.add_argument('--get', '-g',
action='store_true',
help='Ignored; for compatibility with older scripts.')
parser.add_argument('--set', '-s',
action='store_true',
help='Ignored; for compatibility with older scripts.')
parser.add_argument('property', metavar='PROPERTY',
nargs='?',
help='name of the property to show or change')
parser_value = parser.add_mutually_exclusive_group()
parser_value.add_argument('value', metavar='VALUE',
nargs='?',
help='new value of the property')
parser.add_argument('--unset', '--default', '--delete', '-D',
dest='delete',
action='store_true',
help='unset the property; if property has default value, it will be used'
' instead')
def main(args=None):
args = parser.parse_args(args)
args.domain = args.domains.pop()
if args.property is None:
properties = args.domain.property_list()
width = max(len(prop.__name__) for prop in properties)
for prop in sorted(properties):
try:
value = getattr(args.domain, prop.__name__)
except AttributeError:
print('{name:{width}s} U'.format(
name=prop.__name__, width=width))
continue
if args.domain.property_is_default(prop):
print('{name:{width}s} D {value!s}'.format(
name=prop.__name__, width=width, value=value))
else:
print('{name:{width}s} - {value!s}'.format(
name=prop.__name__, width=width, value=value))
return 0
else:
args.property = args.property.replace('-', '_')
if args.property not in [prop.__name__
for prop in args.domain.property_list()]:
parser.error('no such property: {!r}'.format(args.property))
if args.value is not None:
setattr(args.domain, args.property, args.value)
args.app.save()
return 0
if args.delete:
delattr(args.domain, args.property)
args.app.save()
return 0
print(str(getattr(args.domain, args.property)))
return 0
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,54 +0,0 @@
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2016 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
#
# 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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
''' Remove domains from the system '''
from __future__ import print_function
import sys
from qubes.tools import QubesArgumentParser
parser = QubesArgumentParser(description=__doc__,
want_app=True,
want_force_root=True,
vmname_nargs='+')
parser.add_argument('--just-db',
action='store_true',
help='Remove only from db, don\'t remove files')
def main(args=None): # pylint: disable=missing-docstring
args = parser.parse_args(args)
for vm in args.domains:
del args.app.domains[vm.qid]
args.app.save()
if not args.just_db:
vm.remove_from_disk()
else:
# normally it is done by vm.remove_from_disk(), but it isn't
# called in this case
vm.libvirt_domain.undefine()
return 0
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,140 +0,0 @@
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2010-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2015 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
from __future__ import print_function
import os
import sys
import qubes
import qubes.exc
import qubes.tools
parser = qubes.tools.QubesArgumentParser(vmname_nargs='+')
parser.add_argument('--user', '-u', metavar='USER',
help='run command in a qube as USER')
parser.add_argument('--autostart', '--auto', '-a',
action='store_true', default=False,
help='start the qube if it is not running')
parser.add_argument('--pass-io', '-p',
action='store_true', dest='passio', default=False,
help='pass stdio from remote program')
parser.add_argument('--localcmd', metavar='COMMAND',
help='with --pass-io, pass stdio to the given program')
parser.add_argument('--gui',
action='store_true', default=True,
help='run the command with GUI (default on)')
parser.add_argument('--no-gui', '--nogui',
action='store_false', dest='gui',
help='run the command without GUI')
parser.add_argument('--colour-output', '--color-output', metavar='COLOUR',
action='store', dest='color_output', default=None,
help='mark the qube output with given ANSI colour (ie. "31" for red)')
parser.add_argument('--colour-stderr', '--color-stderr', metavar='COLOUR',
action='store', dest='color_stderr', default=None,
help='mark the qube stderr with given ANSI colour (ie. "31" for red)')
parser.add_argument('--no-colour-output', '--no-color-output',
action='store_false', dest='color_output',
help='disable colouring the stdio')
parser.add_argument('--no-colour-stderr', '--no-color-stderr',
action='store_false', dest='color_stderr',
help='disable colouring the stderr')
parser.add_argument('--filter-escape-chars',
action='store_true', dest='filter_esc',
default=os.isatty(sys.stdout.fileno()),
help='filter terminal escape sequences (default if output is terminal)')
parser.add_argument('--no-filter-escape-chars',
action='store_false', dest='filter_esc',
help='do not filter terminal escape sequences; DANGEROUS when output is a'
' terminal emulator')
parser.add_argument('cmd', metavar='COMMAND',
help='command to run')
def main(args=None):
args = parser.parse_args(args)
if args.color_output is None and args.filter_esc:
args.color_output = '31'
if args.color_output is None and os.isatty(sys.stderr.fileno()):
args.color_stderr = 31
if len(args.domains) > 1 and args.passio:
parser.error('--passio cannot be used when more than 1 qube is chosen')
if args.localcmd and not args.passio:
parser.error('--localcmd have no effect without --pass-io')
if args.color_output and not args.filter_esc:
parser.error('--color-output must be used with --filter-escape-chars')
retcode = 0
for vm in args.domains:
if args.autostart and not vm.is_running():
vm.start()
if args.color_output:
sys.stdout.write('\033[0;{}m'.format(args.color_output))
sys.stdout.flush()
if args.color_stderr:
sys.stderr.write('\033[0;{}m'.format(args.color_stderr))
sys.stderr.flush()
try:
retcode = max(retcode, vm.run(args.cmd,
user=args.user,
passio=args.passio,
localcmd=args.localcmd,
gui=args.gui,
filter_esc=args.filter_esc))
except qubes.exc.QubesException as e:
if args.color_output:
sys.stdout.write('\033[0m')
sys.stdout.flush()
vm.log.error(str(e))
return -1
finally:
if args.color_output:
sys.stdout.write('\033[0m')
sys.stdout.flush()
if args.color_stderr:
sys.stderr.write('\033[0m')
sys.stderr.flush()
return retcode
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,83 +0,0 @@
#
# The Qubes OS Project, http://www.qubes-os.org
#
# Copyright (C) 2010-2016 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2011-2016 Marek Marczykowski-Górecki
# <marmarek@invisiblethingslab.com>
# Copyright (C) 2016 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
''' Shutdown a qube '''
from __future__ import print_function
import sys
import time
import qubes.config
import qubes.tools
parser = qubes.tools.QubesArgumentParser(
description=__doc__, vmname_nargs='+')
parser.add_argument('--force',
action='store_true', default=False,
help='force operation, even if may damage other VMs (eg. shutdown of'
' network provider)')
parser.add_argument('--wait',
action='store_true', default=False,
help='wait for the VMs to shut down')
parser.add_argument('--timeout',
action='store', type=float,
default=qubes.config.defaults['shutdown_counter_max'],
help='timeout after which domains are killed when using --wait'
' (default: %d)')
def main(args=None): # pylint: disable=missing-docstring
args = parser.parse_args(args)
for vm in args.domains:
if not vm.is_halted():
vm.shutdown(force=args.force)
if not args.wait:
return
timeout = args.timeout
current_vms = list(sorted(args.domains))
while timeout >= 0:
current_vms = [vm for vm in current_vms
if vm.get_power_state() != 'Halted']
if not current_vms:
return 0
args.app.log.info('Waiting for shutdown ({}): {}'.format(
timeout, ', '.join([str(vm) for vm in current_vms])))
time.sleep(1)
timeout -= 1
args.app.log.info(
'Killing remaining qubes: {}'
.format(', '.join([str(vm) for vm in current_vms])))
for vm in current_vms:
vm.force_shutdown()
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,152 +0,0 @@
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2010-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2015 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
'''qvm-start - Start a domain'''
# TODO notification in tray
import argparse
import os
import sys
import qubes
class DriveAction(argparse.Action):
'''Action for argument parser that stores drive image path.'''
# pylint: disable=redefined-builtin,too-few-public-methods
def __init__(self,
option_strings,
dest='drive',
prefix='cdrom:',
metavar='IMAGE',
required=False,
help='Attach drive'):
super(DriveAction, self).__init__(option_strings, dest,
metavar=metavar, help=help)
self.prefix = prefix
def __call__(self, parser, namespace, values, option_string=None):
# pylint: disable=redefined-outer-name
setattr(namespace, self.dest, self.prefix + values)
parser = qubes.tools.QubesArgumentParser(vmname_nargs=1,
description='start a domain')
parser_drive = parser.add_mutually_exclusive_group()
parser_drive.add_argument('--drive', metavar='DRIVE',
help='temporarily attach specified drive as CD/DVD or hard disk (can be'
' specified with prefix "hd:" or "cdrom:", default is cdrom)')
parser_drive.add_argument('--hddisk',
action=DriveAction, prefix='hd:',
help='temporarily attach specified drive as hard disk')
parser_drive.add_argument('--cdrom', metavar='IMAGE',
action=DriveAction, prefix='cdrom:',
help='temporarily attach specified drive as CD/DVD')
parser_drive.add_argument('--install-windows-tools',
action='store_const', dest='drive', default=False,
const='cdrom:dom0:/usr/lib/qubes/qubes-windows-tools.iso',
help='temporarily attach Windows tools CDROM to the domain')
parser.add_argument('--conf-file', metavar='FILE',
help='use custom libvirt config instead of Qubes-generated one')
parser.add_argument('--debug',
action='store_true', default=False,
help='enable debug mode for this domain (until its shutdown)')
parser.add_argument('--preparing-dvm',
action='store_true', default=False,
help='do actions necessary when preparing DVM image')
parser.add_argument('--no-start-guid',
action='store_false', dest='start_guid', default=True,
help='do not start the gui daemon')
parser.add_argument('--no-guid',
action='store_false', dest='start_guid',
help='same as --no-start-guid')
parser.add_argument('--skip-if-running',
action='store_true', default=False,
help='Do not fail if the qube is already runnning')
#parser.add_option ("--tray", action="store_true", dest="tray", default=False,
# help="Use tray notifications instead of stdout" )
parser.set_defaults(drive=None)
def main(args=None):
'''Main routine of :program:`qvm-start`.
:param list args: Optional arguments to override those delivered from \
command line.
'''
args = parser.parse_args(args)
# if options.tray:
# tray_notify_init()
vm = args.domains[0]
if args.skip_if_running and vm.is_running():
return
if args.drive is not None:
if 'drive' not in (prop.__name__ for prop in vm.property_list()):
parser.error(
'domain {!r} does not support attaching drives'.format(vm.name))
else:
if args.drive == 'cdrom:dom0:/usr/lib/qubes/qubes-windows-tools.iso':
path = args.drive.split(':', 2)[2]
if not os.path.exists(path):
parser.error('qubes-windows-tools package not installed')
if args.conf_file is not None:
vm.conf_file = args.conf_file
if args.debug:
vm.debug = args.debug
if args.debug:
vm.start(
preparing_dvm=args.preparing_dvm,
start_guid=args.start_guid)
else:
try:
vm.start(
preparing_dvm=args.preparing_dvm,
start_guid=args.start_guid)
except qubes.exc.QubesException as e:
parser.error_runtime('Qubes error: {!r}'.format(e))
return 0
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,95 +0,0 @@
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2010-2016 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2016 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
'''qvm-features - Manage domain's tags'''
from __future__ import print_function
import sys
import qubes
parser = qubes.tools.QubesArgumentParser(
vmname_nargs=1,
description='manage domain\'s tags')
mode = parser.add_mutually_exclusive_group()
def mode_query(args):
if args.tag is None:
# list
print('\n'.join(sorted(args.vm.tags)))
else:
# real query; logic is inverted, because this is exit code
return int(args.tag not in args.vm.tags)
mode.add_argument('--query',
dest='mode',
action='store_const',
const=mode_query,
help='query for the tag; if no tag specified, list all tags;'
' this is the default')
def mode_set(args):
if args.tag is None:
parser.error('tag is mandatory for --set')
args.vm.tags.add(args.tag)
args.app.save()
mode.add_argument('--set', '-s',
dest='mode',
action='store_const',
const=mode_set,
help='set the tag; if tag is already set, do nothing')
def mode_unset(args):
if args.tag is None:
parser.error('tag is mandatory for --unset')
args.vm.tags.discard(args.tag)
args.app.save()
mode.add_argument('--unset', '--delete', '-D',
dest='mode',
action='store_const',
const=mode_unset,
help='unset the tag; if tag is not set, do nothing')
parser.add_argument('tag', metavar='TAG',
action='store', nargs='?',
help='name of the tag')
parser.set_defaults(mode=mode_query)
def main(args=None):
'''Main routine of :program:`qvm-tags`.
:param list args: Optional arguments to override those delivered from \
command line.
'''
args = parser.parse_args(args)
return args.mode(args)
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,49 +0,0 @@
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2010-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2015 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
'''qvm-template-commit - Commit disk changes'''
import sys
import qubes
parser = qubes.tools.QubesArgumentParser(vmname_nargs=1,
description='commit VM disk changes; this tool isn\'t intended to manual '
'use')
def main(args=None):
'''Main routine of :program:`qvm-pause`.
:param list args: Optional arguments to override those delivered from \
command line.
'''
args = parser.parse_args(args)
args.app.offline_mode = True
for domain in args.domains:
domain.storage.commit()
return 0
if __name__ == '__main__':
sys.exit(main())

View File

@ -1,48 +0,0 @@
#
# The Qubes OS Project, https://www.qubes-os.org/
#
# Copyright (C) 2010-2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
# Copyright (C) 2015 Wojtek Porczyk <woju@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, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
'''qvm-unpause - Unpause a domain'''
import sys
import qubes
parser = qubes.tools.QubesArgumentParser(
vmname_nargs='+',
description='unpause a domain')
def main(args=None):
'''Main routine of :program:`qvm-unpause`.
:param list args: Optional arguments to override those delivered from \
command line.
'''
args = parser.parse_args(args)
for domain in args.domains:
domain.unpause()
return 0
if __name__ == '__main__':
sys.exit(main())

View File

@ -273,30 +273,11 @@ fi
%{python3_sitelib}/qubes/tools/__init__.py
%{python3_sitelib}/qubes/tools/qmemmand.py
%{python3_sitelib}/qubes/tools/qubes_create.py
%{python3_sitelib}/qubes/tools/qubes_prefs.py
%{python3_sitelib}/qubes/tools/qubesd.py
%{python3_sitelib}/qubes/tools/qubesd_query.py
%{python3_sitelib}/qubes/tools/qvm_block.py
%{python3_sitelib}/qubes/tools/qvm_backup.py
%{python3_sitelib}/qubes/tools/qvm_backup_restore.py
%{python3_sitelib}/qubes/tools/qvm_create.py
%{python3_sitelib}/qubes/tools/qvm_device.py
%{python3_sitelib}/qubes/tools/qvm_features.py
%{python3_sitelib}/qubes/tools/qvm_firewall.py
%{python3_sitelib}/qubes/tools/qvm_check.py
%{python3_sitelib}/qubes/tools/qvm_clone.py
%{python3_sitelib}/qubes/tools/qvm_kill.py
%{python3_sitelib}/qubes/tools/qvm_pause.py
%{python3_sitelib}/qubes/tools/qvm_pool.py
%{python3_sitelib}/qubes/tools/qvm_prefs.py
%{python3_sitelib}/qubes/tools/qvm_remove.py
%{python3_sitelib}/qubes/tools/qvm_run.py
%{python3_sitelib}/qubes/tools/qvm_shutdown.py
%{python3_sitelib}/qubes/tools/qvm_start.py
%{python3_sitelib}/qubes/tools/qvm_tags.py
%{python3_sitelib}/qubes/tools/qvm_template_commit.py
%{python3_sitelib}/qubes/tools/qvm_template_postprocess.py
%{python3_sitelib}/qubes/tools/qvm_unpause.py
%dir %{python3_sitelib}/qubes/ext
%dir %{python3_sitelib}/qubes/ext/__pycache__
@ -343,10 +324,6 @@ fi
%dir %{python3_sitelib}/qubes/tests/tools/__pycache__
%{python3_sitelib}/qubes/tests/tools/__pycache__/*
%{python3_sitelib}/qubes/tests/tools/__init__.py
%{python3_sitelib}/qubes/tests/tools/init.py
%{python3_sitelib}/qubes/tests/tools/qubesd.py
%{python3_sitelib}/qubes/tests/tools/qvm_device.py
%{python3_sitelib}/qubes/tests/tools/qvm_firewall.py
%{python3_sitelib}/qubes/tests/tools/qubesd.py
%dir %{python3_sitelib}/qubes/tests/integ
@ -368,11 +345,6 @@ fi
%{python3_sitelib}/qubes/tests/integ/tools/__pycache__/*
%{python3_sitelib}/qubes/tests/integ/tools/__init__.py
%{python3_sitelib}/qubes/tests/integ/tools/qubes_create.py
%{python3_sitelib}/qubes/tests/integ/tools/qvm_features.py*
%{python3_sitelib}/qubes/tests/integ/tools/qvm_firewall.py
%{python3_sitelib}/qubes/tests/integ/tools/qvm_check.py
%{python3_sitelib}/qubes/tests/integ/tools/qvm_prefs.py
%{python3_sitelib}/qubes/tests/integ/tools/qvm_run.py
%dir %{python3_sitelib}/qubes/qmemman
%dir %{python3_sitelib}/qubes/qmemman/__pycache__