Use qubes.VMExec in run_with_args()

This commit is contained in:
Pawel Marczewski 2020-01-27 10:24:22 +01:00
parent 8416c033c8
commit 2597ed0795
No known key found for this signature in database
GPG Key ID: DE42EE9B14F96465
3 changed files with 35 additions and 3 deletions

View File

@ -189,6 +189,10 @@ class TC_00_qvm_start(qubesadmin.tests.QubesTestCase):
('some-vm', 'admin.vm.device.block.Set.persistent',
'other-vm+loop7',
b'False')] = b'0\x00'
self.app.expected_calls[
('other-vm', 'admin.vm.feature.CheckWithTemplate', 'vmexec',
None)] = b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'vmexec\' not set\x00'
with unittest.mock.patch.object(self.app.domains['other-vm'], 'run') \
as mock_run:
mock_run.return_value = (b'/dev/loop7', b'')

View File

@ -1,4 +1,4 @@
# -*- encoding: utf8 -*-
# -*- encoding: utf-8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -94,7 +94,10 @@ class TC_00_Actions(qubesadmin.tests.vm.VMTestCase):
('test-vm', 'qubes.VMShell', b'some command& exit\n'),
])
def test_015_run_with_args(self):
def test_015_run_with_args_shell(self):
self.app.expected_calls[
('test-vm', 'admin.vm.feature.CheckWithTemplate', 'vmexec', None)] = \
b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'vmexec\' not set\x00'
self.app.expected_calls[
('test-vm', 'admin.vm.feature.CheckWithTemplate', 'os', None)] = \
b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'os\' not set\x00'
@ -106,3 +109,18 @@ class TC_00_Actions(qubesadmin.tests.vm.VMTestCase):
b'some \'argument with spaces\' \'and $pecial; chars\'; '
b'exit\n'),
])
def test_016_run_with_args_exec(self):
self.app.expected_calls[
('test-vm', 'admin.vm.feature.CheckWithTemplate', 'vmexec', None)] = \
b'0\x001'
self.vm.run_with_args('some', 'argument with spaces',
'and $pecial; chars')
self.assertEqual(self.app.service_calls, [
('test-vm',
'qubes.VMExec+some+argument-20with-20spaces+and-20-24pecial-3B-20chars',
{}),
('test-vm',
'qubes.VMExec+some+argument-20with-20spaces+and-20-24pecial-3B-20chars',
b''),
])

View File

@ -316,12 +316,22 @@ class QubesVM(qubesadmin.base.PropertyHolder):
raise e
def run_with_args(self, *args, **kwargs):
'''Run a single command inside the domain using qubes.VMShell qrexec.
'''Run a single command inside the domain. Use the qubes.VMExec qrexec,
if available.
This method execute a single command, without interpreting any shell
special characters.
''' # pylint: disable=redefined-builtin
if self.features.check_with_template('vmexec', False):
try:
return self.run_service_for_stdio(
'qubes.VMExec+' + qubesadmin.utils.encode_for_vmexec(args),
**kwargs)
except subprocess.CalledProcessError as e:
e.cmd = str(args)
raise e
return self.run(' '.join(shlex.quote(arg) for arg in args), **kwargs)
@property