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

Merge branch 'devel-20181206'

Browse Source
This commit is contained in:
Marek Marczykowski-Górecki 2018-12-09 18:08:25 +01:00
commit 9061169f90
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
21 changed files with 385 additions and 193 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
doc/index.rst
View File

@ -11,6 +11,9 @@ Contents:
.. toctree::
:maxdepth: 2
modules
manpages/index
Indices and tables

22
doc/manpages/qvm-device.rst
View File

@ -6,13 +6,12 @@
Synopsis
========
| :command:`qvm-device` [*options*] *DEVICE_CLASS* {list,ls,l} <*vm-name*>
| :command:`qvm-device` [*options*] *DEVICE_CLASS* {attach,at,a} <*vm-name*> <*device*>
| :command:`qvm-device` [*options*] *DEVICE_CLASS* {detach,dt,d} <*vm-name*> <*device*>
| :command:`qvm-*DEVICE_CLASS*` [*options*] {list,ls,l,attach,at,a,detach,dt,d} <*vmname*> ...
| :command:`qvm-device` *DEVICE_CLASS* {list,ls,l} [*options*] <*vm-name*>
| :command:`qvm-device` *DEVICE_CLASS* {attach,at,a} [*options*] <*vm-name*> <*device*>
| :command:`qvm-device` *DEVICE_CLASS* {detach,dt,d} [*options*] <*vm-name*> [<*device*>]
| :command:`qvm-*DEVICE_CLASS*` {list,ls,l,attach,at,a,detach,dt,d} [*options*] <*vmname*> ...
Tool can be called either as `qvm-device *DEVICE_CLASS* ...`, or
`qvm-*DEVICE_CLASS* ...`. The latter is used for `qvm-pci`, `qvm-block` etc.
.. note:: :command:`qvm-block`, :command:`qvm-usb` and :command:`qvm-pci` are just aliases for :command:`qvm-device block`, :command:`qvm-device usb` and :command:`qvm-device pci` respectively.
Options
=======
@ -79,7 +78,8 @@ detach
| :command:`qvm-device` *DEVICE_CLASS* detach [-h] [--verbose] [--quiet] *VMNAME* *BACKEND_DOMAIN:DEVICE_ID*
Detach the device with *BACKEND_DOMAIN:DEVICE_ID* from domain *VMNAME*
Detach the device with *BACKEND_DOMAIN:DEVICE_ID* from domain *VMNAME*.
If no device is given, detach all *DEVICE_CLASS* devices.
aliases: d, dt
@ -107,7 +107,15 @@ pci
PCI device. Only dom0 expose such devices. One should be very careful when attaching this type of devices, because some of them are strictly required to stay in dom0 (for example host bridge). Available options:
* `no-strict-reset` - allow to attach device even if it does not support any reliable reset operation; switching such device to another domain (without full host restart) can be a security risk; default: `False`, accepted values: `True`, `False` (option absent)
* `permissive` - allow write access to most of PCI config space, instead of only selected whitelisted rregisters; a workaround for some PCI passthrough problems, potentially unsafe; default: `False`, accepted values: `True`, `False` (option absent)
mic
^^^
Microphone, or other audio input. Normally there is only one device of this
type - `dom0:mic`. Use PulseAudio settings in dom0 to select which input source
is used.
This type of device does not support options.
Authors
=======

17
qubesadmin/app.py
View File

@ -258,6 +258,7 @@ class QubesBase(qubesadmin.base.PropertyHolder):
can be also VM object; use None for default value
:param str pool: storage pool to use instead of default one
:param dict pools: storage pool for specific volumes
:return new VM object
'''
@ -310,6 +311,7 @@ class QubesBase(qubesadmin.base.PropertyHolder):
logged, or abort the whole operation?
:param list ignore_volumes: do not clone volumes on this list,
like 'private' or 'root'
:return new VM object
'''
@ -328,6 +330,21 @@ class QubesBase(qubesadmin.base.PropertyHolder):
label = src_vm.label
if pool is None and pools is None:
# use the same pools as the source - check if non default is used
for volume in sorted(src_vm.volumes.values()):
if not volume.save_on_stop:
# clone only persistent volumes
continue
if ignore_volumes and volume.name in ignore_volumes:
continue
default_pool = getattr(self.app, 'default_pool_' + volume.name,
volume.pool)
if default_pool != volume.pool:
if pools is None:
pools = {}
pools[volume.name] = volume.pool
method_prefix = 'admin.vm.Create.'
payload = 'name={} label={}'.format(new_name, label)
if pool:

12
qubesadmin/tests/__init__.py
View File

@ -17,6 +17,7 @@
#
# You should have received a copy of the GNU Lesser General Public License along
# with this program; if not, see <http://www.gnu.org/licenses/>.
import subprocess
import traceback
import unittest
@ -61,7 +62,13 @@ class TestProcess(object):
lambda: self.input_callback(self.stdin.getvalue()))
else:
self.stdin.close = lambda: None
if stdout == subprocess.PIPE:
self.stdout = io.BytesIO()
else:
self.stdout = stdout
if stderr == subprocess.PIPE:
self.stderr = io.BytesIO()
else:
self.stderr = stderr
self.returncode = 0
@ -146,7 +153,10 @@ class QubesTest(qubesadmin.app.QubesBase):
def run_service(self, dest, service, **kwargs):
self.service_calls.append((dest, service, kwargs))
return TestProcess(lambda input: self.service_calls.append((dest,
service, input)))
service, input)),
stdout=kwargs.get('stdout', None),
stderr=kwargs.get('stderr', None),
)
class QubesTestCase(unittest.TestCase):

65
qubesadmin/tests/app.py
View File

@ -298,13 +298,11 @@ class TC_10_QubesBase(qubesadmin.tests.QubesTestCase):
b'0\x00'
# storage
for vm in (src, dst):
self.app.expected_calls[
(dst, 'admin.vm.volume.List', None, None)] = \
b'0\x00root\nprivate\nvolatile\nkernel\n'
self.app.expected_calls[
(dst, 'admin.vm.volume.Info', 'root', None)] = \
(vm, 'admin.vm.volume.Info', 'root', None)] = \
b'0\x00pool=lvm\n' \
b'vid=vm-test-vm/root\n' \
b'vid=vm-' + vm.encode() + b'/root\n' \
b'size=10737418240\n' \
b'usage=2147483648\n' \
b'rw=False\n' \
@ -313,9 +311,9 @@ class TC_10_QubesBase(qubesadmin.tests.QubesTestCase):
b'save_on_stop=False\n' \
b'snap_on_start=True\n'
self.app.expected_calls[
(dst, 'admin.vm.volume.Info', 'private', None)] = \
(vm, 'admin.vm.volume.Info', 'private', None)] = \
b'0\x00pool=lvm\n' \
b'vid=vm-test-vm/private\n' \
b'vid=vm-' + vm.encode() + b'/private\n' \
b'size=2147483648\n' \
b'usage=214748364\n' \
b'rw=True\n' \
@ -323,9 +321,9 @@ class TC_10_QubesBase(qubesadmin.tests.QubesTestCase):
b'save_on_stop=True\n' \
b'snap_on_start=False\n'
self.app.expected_calls[
(dst, 'admin.vm.volume.Info', 'volatile', None)] = \
(vm, 'admin.vm.volume.Info', 'volatile', None)] = \
b'0\x00pool=lvm\n' \
b'vid=vm-test-vm/volatile\n' \
b'vid=vm-' + vm.encode() + b'/volatile\n' \
b'size=10737418240\n' \
b'usage=0\n' \
b'rw=True\n' \
@ -334,7 +332,7 @@ class TC_10_QubesBase(qubesadmin.tests.QubesTestCase):
b'save_on_stop=False\n' \
b'snap_on_start=False\n'
self.app.expected_calls[
(dst, 'admin.vm.volume.Info', 'kernel', None)] = \
(vm, 'admin.vm.volume.Info', 'kernel', None)] = \
b'0\x00pool=linux-kernel\n' \
b'vid=\n' \
b'size=0\n' \
@ -345,7 +343,7 @@ class TC_10_QubesBase(qubesadmin.tests.QubesTestCase):
b'save_on_stop=False\n' \
b'snap_on_start=False\n'
self.app.expected_calls[
(src, 'admin.vm.volume.List', None, None)] = \
(vm, 'admin.vm.volume.List', None, None)] = \
b'0\x00root\nprivate\nvolatile\nkernel\n'
self.app.expected_calls[
(src, 'admin.vm.volume.CloneFrom', 'private', None)] = \
@ -353,6 +351,9 @@ class TC_10_QubesBase(qubesadmin.tests.QubesTestCase):
self.app.expected_calls[
(dst, 'admin.vm.volume.CloneTo', 'private', b'token-private')] = \
b'0\x00'
self.app.expected_calls[
('dom0', 'admin.property.Get', 'default_pool_private', None)] = \
b'0\0default=True type=str lvm'
def test_030_clone(self):
self.clone_setup_common_calls('test-vm', 'new-name')
@ -387,6 +388,11 @@ class TC_10_QubesBase(qubesadmin.tests.QubesTestCase):
def test_032_clone_pool(self):
self.clone_setup_common_calls('test-vm', 'new-name')
for volume in ('root', 'private', 'volatile', 'kernel'):
del self.app.expected_calls[
'test-vm', 'admin.vm.volume.Info', volume, None]
del self.app.expected_calls[
'dom0', 'admin.property.Get', 'default_pool_private', None]
self.app.expected_calls[('dom0', 'admin.vm.CreateInPool.AppVM',
'test-template',
b'name=new-name label=red pool=some-pool')] = b'0\x00'
@ -401,6 +407,11 @@ class TC_10_QubesBase(qubesadmin.tests.QubesTestCase):
def test_033_clone_pools(self):
self.clone_setup_common_calls('test-vm', 'new-name')
for volume in ('root', 'private', 'volatile', 'kernel'):
del self.app.expected_calls[
'test-vm', 'admin.vm.volume.Info', volume, None]
del self.app.expected_calls[
'dom0', 'admin.property.Get', 'default_pool_private', None]
self.app.expected_calls[('dom0', 'admin.vm.CreateInPool.AppVM',
'test-template',
b'name=new-name label=red pool:private=some-pool '
@ -451,6 +462,10 @@ class TC_10_QubesBase(qubesadmin.tests.QubesTestCase):
self.app.expected_calls[
('test-vm', 'admin.vm.property.List', None, None)] = \
b'0\0qid\nname\ntemplate\nlabel\nmemory\n'
# simplify it a little, shouldn't get this far anyway
self.app.expected_calls[
('test-vm', 'admin.vm.volume.List', None, None)] = \
b'0\x00'
self.app.expected_calls[
('test-vm', 'admin.vm.property.Get', 'label', None)] = \
b'0\0default=False type=label red'
@ -587,6 +602,34 @@ class TC_10_QubesBase(qubesadmin.tests.QubesTestCase):
self.app.clone_vm('test-vm', 'new-name')
self.assertAllCalled()
def test_042_clone_nondefault_pool(self):
self.clone_setup_common_calls('test-vm', 'new-name')
self.app.expected_calls[
('test-vm', 'admin.vm.volume.Info', 'private', None)] = \
b'0\x00pool=another\n' \
b'vid=vm-test-vm/private\n' \
b'size=2147483648\n' \
b'usage=214748364\n' \
b'rw=True\n' \
b'internal=True\n' \
b'save_on_stop=True\n' \
b'snap_on_start=False\n'
self.app.expected_calls[('dom0', 'admin.vm.List', None, None)] = \
b'0\x00new-name class=AppVM state=Halted\n' \
b'test-vm class=AppVM state=Halted\n' \
b'test-template class=TemplateVM state=Halted\n' \
b'test-net class=AppVM state=Halted\n'
self.app.expected_calls[('dom0', 'admin.vm.CreateInPool.AppVM',
'test-template', b'name=new-name label=red pool:private=another')]\
= b'0\x00'
new_vm = self.app.clone_vm('test-vm', 'new-name')
self.assertEqual(new_vm.name, 'new-name')
self.check_output_mock.assert_called_once_with(
['qvm-appmenus', '--init', '--update',
'--source', 'test-vm', 'new-name'],
stderr=subprocess.STDOUT
)
self.assertAllCalled()
class TC_20_QubesLocal(unittest.TestCase):

28
qubesadmin/tests/tools/qvm_create.py
View File

@ -99,7 +99,7 @@ class TC_00_qvm_create(qubesadmin.tests.QubesTestCase):
with tempfile.NamedTemporaryFile() as root_file:
root_file.file.write(b'root data')
root_file.file.flush()
self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
self.app.expected_calls[('dom0', 'admin.vm.Create.StandaloneVM',
None, b'name=new-vm label=red')] = b'0\x00'
self.app.expected_calls[('dom0', 'admin.label.List', None, None)] = \
b'0\x00red\nblue\n'
@ -117,7 +117,7 @@ class TC_00_qvm_create(qubesadmin.tests.QubesTestCase):
self.app.expected_calls[
('new-vm', 'admin.vm.volume.Import', 'root', b'root data')] = \
b'0\0'
qubesadmin.tools.qvm_create.main(['-l', 'red',
qubesadmin.tools.qvm_create.main(['-l', 'red', '-C', 'StandaloneVM',
'--root-copy-from=' + root_file.name, 'new-vm'],
app=self.app)
self.assertAllCalled()
@ -127,7 +127,7 @@ class TC_00_qvm_create(qubesadmin.tests.QubesTestCase):
with tempfile.NamedTemporaryFile(delete=False) as root_file:
root_file.file.write(b'root data')
root_file.file.flush()
self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
self.app.expected_calls[('dom0', 'admin.vm.Create.StandaloneVM',
None, b'name=new-vm label=red')] = b'0\x00'
self.app.expected_calls[('dom0', 'admin.label.List', None, None)] = \
b'0\x00red\nblue\n'
@ -145,7 +145,7 @@ class TC_00_qvm_create(qubesadmin.tests.QubesTestCase):
self.app.expected_calls[
('new-vm', 'admin.vm.volume.Import', 'root', b'root data')] = \
b'0\0'
qubesadmin.tools.qvm_create.main(['-l', 'red',
qubesadmin.tools.qvm_create.main(['-l', 'red', '-C', 'StandaloneVM',
'--root-move-from=' + root_file.name, 'new-vm'],
app=self.app)
self.assertAllCalled()
@ -156,7 +156,7 @@ class TC_00_qvm_create(qubesadmin.tests.QubesTestCase):
root_file.file.write(b'root data')
root_file.file.flush()
with self.assertRaises(SystemExit):
qubesadmin.tools.qvm_create.main(['-l', 'red',
qubesadmin.tools.qvm_create.main(['-l', 'red', '-C', 'StandaloneVM',
'--root-copy-from=' + root_file.name,
'--root-move-from=' + root_file.name,
'new-vm'],
@ -166,7 +166,7 @@ class TC_00_qvm_create(qubesadmin.tests.QubesTestCase):
def test_008_root_invalid_path(self):
with self.assertRaises(SystemExit):
qubesadmin.tools.qvm_create.main(['-l', 'red',
qubesadmin.tools.qvm_create.main(['-l', 'red', '-C', 'StandaloneVM',
'--root-copy-from=/invalid', 'new-vm'],
app=self.app)
self.assertAllCalled()
@ -185,7 +185,7 @@ class TC_00_qvm_create(qubesadmin.tests.QubesTestCase):
with tempfile.NamedTemporaryFile() as root_file:
root_file.file.write(b'root data')
root_file.file.flush()
self.app.expected_calls[('dom0', 'admin.vm.Create.AppVM',
self.app.expected_calls[('dom0', 'admin.vm.Create.StandaloneVM',
None, b'name=new-vm label=red')] = b'0\x00'
self.app.expected_calls[('dom0', 'admin.label.List', None, None)] = \
b'0\x00red\nblue\n'
@ -206,7 +206,7 @@ class TC_00_qvm_create(qubesadmin.tests.QubesTestCase):
self.app.expected_calls[
('new-vm', 'admin.vm.volume.Import', 'root', b'root data')] = \
b'0\0'
qubesadmin.tools.qvm_create.main(['-l', 'red',
qubesadmin.tools.qvm_create.main(['-l', 'red', '-C', 'StandaloneVM',
'--root-copy-from=' + root_file.name, 'new-vm'],
app=self.app)
self.assertAllCalled()
@ -329,6 +329,18 @@ class TC_00_qvm_create(qubesadmin.tests.QubesTestCase):
self.assertIn('red, blue', stderr.getvalue())
self.assertAllCalled()
def test_013_root_copy_from_template_based(self):
with tempfile.NamedTemporaryFile() as root_file:
root_file.file.write(b'root data')
root_file.file.flush()
with self.assertRaises(SystemExit):
with qubesadmin.tests.tools.StderrBuffer() as stderr:
qubesadmin.tools.qvm_create.main(['-l', 'red',
'--root-copy-from=' + root_file.name, 'new-vm'],
app=self.app)
self.assertIn('--root-copy-from', stderr.getvalue())
self.assertAllCalled()
def test_014_standalone_shortcut(self):
self.app.expected_calls[('dom0', 'admin.vm.Create.StandaloneVM',
None, b'name=new-vm label=red')] = b'0\x00'

13
qubesadmin/tests/tools/qvm_device.py
View File

@ -199,3 +199,16 @@ class TC_00_qvm_device(qubesadmin.tests.QubesTestCase):
['test', 'detach', 'test-vm2', 'test-vm1:dev7'], app=self.app)
self.assertAllCalled()
def test_022_detach_all(self):
''' Test detach action '''
self.app.expected_calls[('test-vm2', 'admin.vm.device.test.List',
None, None)] = \
b'0\0test-vm1+dev1\ntest-vm1+dev2\n'
self.app.expected_calls[('test-vm2', 'admin.vm.device.test.Detach',
'test-vm1+dev1', None)] = b'0\0'
self.app.expected_calls[('test-vm2', 'admin.vm.device.test.Detach',
'test-vm1+dev2', None)] = b'0\0'
qubesadmin.tools.qvm_device.main(
['test', 'detach', 'test-vm2'], app=self.app)
self.assertAllCalled()

80
qubesadmin/tests/tools/qvm_run.py
View File

@ -17,7 +17,6 @@
#
# You should have received a copy of the GNU Lesser General Public License along
# with this program; if not, see <http://www.gnu.org/licenses/>.
import io
import os
import unittest.mock
@ -55,7 +54,6 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('test-vm', 'qubes.VMShell', {
'localcmd': None,
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
@ -91,14 +89,12 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('test-vm', 'qubes.VMShell', {
'localcmd': None,
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
}),
('test-vm', 'qubes.VMShell', b'command; exit\n'),
('test-vm2', 'qubes.VMShell', {
'localcmd': None,
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
@ -107,7 +103,6 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
])
self.assertAllCalled()
@unittest.expectedFailure
def test_002_passio(self):
self.app.expected_calls[
('dom0', 'admin.vm.List', None, None)] = \
@ -121,19 +116,51 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
echo = subprocess.Popen(['echo', 'some-data'], stdout=subprocess.PIPE)
with unittest.mock.patch('sys.stdin', echo.stdout):
ret = qubesadmin.tools.qvm_run.main(
['--no-gui', '--pass-io', 'test-vm', 'command'],
['--no-gui', '--pass-io', '--filter-escape-chars',
'test-vm', 'command'],
app=self.app)
echo.stdout.close()
echo.wait()
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('test-vm', 'qubes.VMShell', {
'filter_esc': self.default_filter_esc(),
'localcmd': None,
'filter_esc': True,
'stdout': None,
'stderr': None,
'user': None,
}),
('test-vm', 'qubes.VMShell', b'command; exit\nsome-data\n')
# TODO: find a way to compare b'some-data\n' sent from another
# proces
('test-vm', 'qubes.VMShell', b'command; exit\n')
])
self.assertAllCalled()
def test_002_passio_service(self):
self.app.expected_calls[
('dom0', 'admin.vm.List', None, None)] = \
b'0\x00test-vm class=AppVM state=Running\n'
# self.app.expected_calls[
# ('test-vm', 'admin.vm.List', None, None)] = \
# b'0\x00test-vm class=AppVM state=Running\n'
echo = subprocess.Popen(['echo', 'some-data'], stdout=subprocess.PIPE)
with unittest.mock.patch('sys.stdin', echo.stdout):
ret = qubesadmin.tools.qvm_run.main(
['--no-gui', '--service', '--pass-io', '--filter-escape-chars',
'test-vm', 'test.service'],
app=self.app)
echo.stdout.close()
echo.wait()
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('test-vm', 'test.service', {
'filter_esc': True,
'stdout': None,
'stderr': None,
'user': None,
}),
# TODO: find a way to compare b'some-data\n' sent from another
# proces
('test-vm', 'test.service', b'')
])
self.assertAllCalled()
@ -156,12 +183,12 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
['--no-gui', '--filter-esc', '--pass-io', 'test-vm',
'command'],
app=self.app)
echo.stdout.close()
echo.wait()
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('test-vm', 'qubes.VMShell', {
'filter_esc': True,
'localcmd': None,
'stdout': None,
'stderr': None,
'user': None,
@ -189,11 +216,12 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
'test-vm', 'command'],
app=self.app)
echo.stdout.close()
echo.wait()
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('test-vm', 'qubes.VMShell', {
'filter_esc': self.default_filter_esc(),
'localcmd': None,
'stdout': None,
'stderr': None,
'user': None,
@ -224,11 +252,12 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
'test-vm', 'command'],
app=self.app)
echo.stdout.close()
echo.wait()
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('test-vm', 'qubes.VMShell', {
'filter_esc': False,
'localcmd': None,
'stdout': None,
'stderr': None,
'user': None,
@ -239,7 +268,8 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
stdout.close()
self.assertAllCalled()
def test_005_localcmd(self):
@unittest.mock.patch('subprocess.Popen')
def test_005_localcmd(self, mock_popen):
self.app.expected_calls[
('dom0', 'admin.vm.List', None, None)] = \
b'0\x00test-vm class=AppVM state=Running\n'
@ -249,6 +279,7 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
# self.app.expected_calls[
# ('test-vm', 'admin.vm.List', None, None)] = \
# b'0\x00test-vm class=AppVM state=Running\n'
mock_popen.return_value.wait.return_value = 0
ret = qubesadmin.tools.qvm_run.main(
['--no-gui', '--pass-io', '--localcmd', 'local-command',
'test-vm', 'command'],
@ -256,13 +287,16 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('test-vm', 'qubes.VMShell', {
'localcmd': 'local-command',
'stdout': None,
'stdout': subprocess.PIPE,
'stdin': subprocess.PIPE,
'stderr': None,
'user': None,
}),
('test-vm', 'qubes.VMShell', b'command; exit\n')
])
mock_popen.assert_called_once_with('local-command',
# TODO: check if the right stdin/stdout objects are used
stdout=unittest.mock.ANY, stdin=unittest.mock.ANY, shell=True)
self.assertAllCalled()
def test_006_run_single_with_gui(self):
@ -290,7 +324,6 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
}),
('test-vm', 'qubes.WaitForSession', b'user'),
('test-vm', 'qubes.VMShell', {
'localcmd': None,
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
@ -321,7 +354,6 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
}),
('test-vm', 'qubes.WaitForSession', b'user'),
('test-vm', 'service.name', {
'localcmd': None,
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
@ -336,7 +368,6 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('$dispvm', 'test.service', {
'localcmd': None,
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
@ -351,7 +382,6 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('$dispvm:test-vm', 'test.service', {
'localcmd': None,
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
@ -375,7 +405,6 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('disp123', 'test.service', {
'localcmd': None,
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
@ -400,7 +429,6 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('disp123', 'test.service', {
'localcmd': None,
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
@ -431,7 +459,6 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('test-vm', 'qubes.VMShell', {
'localcmd': None,
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
@ -452,7 +479,7 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
ret = qubesadmin.tools.qvm_run.main(
['--no-gui', '--no-autostart', 'test-vm3', 'command'],
app=self.app)
self.assertEqual(ret, 0)
self.assertEqual(ret, 1)
self.assertEqual(self.app.service_calls, [])
self.assertAllCalled()
@ -474,7 +501,6 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('disp123', 'qubes.VMShell+WaitForSession', {
'localcmd': None,
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
@ -503,7 +529,6 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('disp123', 'qubes.VMShell', {
'localcmd': None,
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
@ -529,7 +554,6 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('test-vm', 'qubes.VMShell', {
'localcmd': None,
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,

11
qubesadmin/tools/__init__.py
View File

@ -327,10 +327,11 @@ class QubesArgumentParser(argparse.ArgumentParser):
:py:class:`qubes.Qubes` object, just add argument for custom xml file
:param mixed vmname_nargs: The number of ``VMNAME`` arguments that should be
consumed. Values include:
- N (an integer) consumes N arguments (and produces a list)
- '?' consumes zero or one arguments
- '*' consumes zero or more arguments (and produces a list)
- '+' consumes one or more arguments (and produces a list)
* N (an integer) consumes N arguments (and produces a list)
* '?' consumes zero or one arguments
* '*' consumes zero or more arguments (and produces a list)
* '+' consumes one or more arguments (and produces a list)
*kwargs* are passed to :py:class:`argparser.ArgumentParser`.
Currenty supported options:
@ -443,7 +444,7 @@ class QubesArgumentParser(argparse.ArgumentParser):
class SubParsersHelpAction(argparse._HelpAction):
''' Print help for all options _and all subparsers_ '''
''' Print help for all options and all subparsers '''
# source https://stackoverflow.com/a/24122778
# pylint: disable=protected-access,too-few-public-methods

4
qubesadmin/tools/dochelpers.py
View File

@ -138,7 +138,7 @@ class OptionsCheckVisitor(docutils.nodes.SparseNodeVisitor):
While the documentation talks about a
'SparseNodeVisitor.depart_document()' function, this function does
not exists. (For details see implementation of
:py:method:`NodeVisitor.dispatch_departure()`) So we need to
:py:meth:`NodeVisitor.dispatch_departure()`) So we need to
manually call this.
'''
if ignored_options is None:
@ -205,7 +205,7 @@ class CommandCheckVisitor(docutils.nodes.SparseNodeVisitor):
While the documentation talks about a
'SparseNodeVisitor.depart_document()' function, this function does
not exists. (For details see implementation of
:py:method:`NodeVisitor.dispatch_departure()`) So we need to
:py:meth:`NodeVisitor.dispatch_departure()`) So we need to
manually call this.
'''
if self.sub_commands:

6
qubesadmin/tools/qvm_create.py
View File

@ -139,6 +139,12 @@ def main(args=None, app=None):
parser.error(
'File pointed by --root-copy-from/--root-move-from does not exist')
# those are known of non-persistent root, do not list those with known
# persistent root, as an extension may add new classes
if root_source_path and args.cls in ('AppVM', 'DispVM'):
parser.error('--root-copy-from/--root-move-from used but this qube '
'does not have own \'root\' volume (uses template\'s one)')
try:
args.app.get_label(args.properties['label'])
except KeyError:

9
qubesadmin/tools/qvm_device.py
View File

@ -130,8 +130,11 @@ 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]
if args.device_assignment:
vm.devices[args.devclass].detach(args.device_assignment)
else:
for device_assignment in vm.devices[args.devclass].assignments():
vm.devices[args.devclass].detach(device_assignment)
@ -169,6 +172,8 @@ class DeviceAction(qubesadmin.tools.QubesAction):
app = namespace.app
backend_device_id = getattr(namespace, self.dest)
devclass = namespace.devclass
if backend_device_id is None:
return
try:
vmname, device_id = backend_device_id.split(':', 1)
@ -232,7 +237,7 @@ def get_parser(device_class=None):
dest='device_assignment',
action=DeviceAction)
detach_parser.add_argument(metavar='BACKEND:DEVICE_ID',
dest='device_assignment',
dest='device_assignment', nargs=argparse.OPTIONAL,
action=DeviceAction, allow_unknown=True)
attach_parser.add_argument('--option', '-o', action='append',

1
qubesadmin/tools/qvm_pool.py
View File

@ -104,7 +104,6 @@ class _Set(qubesadmin.tools.PoolsAction):
'modifications)')
def __call__(self, parser, namespace, name, option_string=None):
print('dupa')
setattr(namespace, 'command', 'set')
super(_Set, self).__call__(parser, namespace, name, option_string)

151
qubesadmin/tools/qvm_run.py
View File

@ -19,8 +19,9 @@
# with this program; if not, see <http://www.gnu.org/licenses/>.
''' qvm-run tool'''
import contextlib
import os
import signal
import subprocess
import sys
@ -115,6 +116,7 @@ def copy_stdin(stream):
# multiprocessing.Process have sys.stdin connected to /dev/null, use fd 0
# directly
while True:
try:
# select so this code works even if fd 0 is non-blocking
select.select([0], [], [])
data = os.read(0, 65536)
@ -122,8 +124,76 @@ def copy_stdin(stream):
break
stream.write(data)
stream.flush()
except KeyboardInterrupt:
break
stream.close()
def print_no_color(msg, file, color):
'''Print a *msg* to *file* without coloring it.
Namely reset to base color first, print a message, then restore color.
'''
if color:
print('\033[0m{}\033[0;{}m'.format(msg, color), file=file)
else:
print(msg, file=file)
def run_command_single(args, vm):
'''Handle a single VM to run the command in'''
run_kwargs = {}
if not args.passio:
run_kwargs['stdout'] = subprocess.DEVNULL
run_kwargs['stderr'] = subprocess.DEVNULL
elif args.localcmd:
run_kwargs['stdin'] = subprocess.PIPE
run_kwargs['stdout'] = subprocess.PIPE
run_kwargs['stderr'] = None
else:
# connect process output to stdout/err directly if --pass-io is given
run_kwargs['stdout'] = None
run_kwargs['stderr'] = None
if args.filter_esc:
run_kwargs['filter_esc'] = True
if isinstance(args.app, qubesadmin.app.QubesLocal) and \
not args.passio and \
not args.localcmd and \
args.service and \
not args.dispvm:
# wait=False works only in dom0; but it's still useful, to save on
# simultaneous vchan connections
run_kwargs['wait'] = False
copy_proc = None
local_proc = None
if args.service:
service = args.cmd
else:
service = 'qubes.VMShell'
if args.gui and args.dispvm:
service += '+WaitForSession'
proc = vm.run_service(service,
user=args.user,
**run_kwargs)
if not args.service:
proc.stdin.write(vm.prepare_input_for_vmshell(args.cmd))
proc.stdin.flush()
if args.localcmd:
local_proc = subprocess.Popen(args.localcmd,
shell=True,
stdout=proc.stdin,
stdin=proc.stdout)
# stdin is closed below
proc.stdout.close()
elif args.passio:
copy_proc = multiprocessing.Process(target=copy_stdin,
args=(proc.stdin,))
copy_proc.start()
# keep the copying process running
proc.stdin.close()
return proc, copy_proc, local_proc
def main(args=None, app=None):
'''Main function of qvm-run tool'''
args = parser.parse_args(args, app=app)
@ -142,25 +212,6 @@ def main(args=None, app=None):
parser.error('--color-output must be used with --filter-escape-chars')
retcode = 0
run_kwargs = {}
if not args.passio:
run_kwargs['stdout'] = subprocess.DEVNULL
run_kwargs['stderr'] = subprocess.DEVNULL
else:
# connect process output to stdout/err directly if --pass-io is given
run_kwargs['stdout'] = None
run_kwargs['stderr'] = None
if not args.localcmd and args.filter_esc:
run_kwargs['filter_esc'] = True
if isinstance(args.app, qubesadmin.app.QubesLocal) and \
not args.passio and \
not args.localcmd and \
args.service and \
not args.dispvm:
# wait=False works only in dom0; but it's still useful, to save on
# simultaneous vchan connections
run_kwargs['wait'] = False
verbose = args.verbose - args.quiet
if args.passio:
@ -189,49 +240,49 @@ def main(args=None, app=None):
procs = []
for vm in domains:
if not args.autostart and not vm.is_running():
if verbose > 0:
print_no_color('Qube \'{}\' not started'.format(vm.name),
file=sys.stderr, color=args.color_stderr)
retcode = max(retcode, 1)
continue
try:
if verbose > 0:
if args.color_output:
print('\033[0mRunning \'{}\' on {}\033[0;{}m'.format(
args.cmd, vm.name, args.color_stderr),
file=sys.stderr)
else:
print('Running \'{}\' on {}'.format(args.cmd, vm.name),
file=sys.stderr)
print_no_color(
'Running \'{}\' on {}'.format(args.cmd, vm.name),
file=sys.stderr, color=args.color_stderr)
if args.gui and not args.dispvm:
wait_session = vm.run_service('qubes.WaitForSession',
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
try:
wait_session.communicate(vm.default_user.encode())
if args.service:
proc = vm.run_service(args.cmd,
user=args.user,
localcmd=args.localcmd,
**run_kwargs)
else:
service = 'qubes.VMShell'
if args.gui and args.dispvm:
service += '+WaitForSession'
proc = vm.run_service(service,
user=args.user,
localcmd=args.localcmd,
**run_kwargs)
proc.stdin.write(vm.prepare_input_for_vmshell(args.cmd))
proc.stdin.flush()
if args.passio and not args.localcmd:
copy_proc = multiprocessing.Process(target=copy_stdin,
args=(proc.stdin,))
copy_proc.start()
# keep the copying process running
proc.stdin.close()
procs.append(proc)
except KeyboardInterrupt:
with contextlib.suppress(ProcessLookupError):
wait_session.send_signal(signal.SIGINT)
break
proc, copy_proc, local_proc = run_command_single(args, vm)
procs.append((vm, proc))
if local_proc:
procs.append((vm, local_proc))
except qubesadmin.exc.QubesException as e:
if args.color_output:
sys.stdout.write('\033[0m')
sys.stdout.flush()
vm.log.error(str(e))
return -1
for proc in procs:
try:
for vm, proc in procs:
this_retcode = proc.wait()
if this_retcode and args.verbose > 0:
print_no_color(
'{}: command failed with code: {}'.format(
vm.name, this_retcode),
file=sys.stderr, color=args.color_stderr)
retcode = max(retcode, proc.wait())
except KeyboardInterrupt:
for vm, proc in procs:
with contextlib.suppress(ProcessLookupError):
proc.send_signal(signal.SIGINT)
for vm, proc in procs:
retcode = max(retcode, proc.wait())
finally:
if dispvm: