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

qvm-run: use qubes.VMExec, if available

Browse Source 
See QubesOS/qubes-issues#4850.
This commit is contained in:
Pawel Marczewski 2020-01-24 18:40:11 +01:00
parent 47188a38e5
commit ff9b81cc3e
No known key found for this signature in database
GPG Key ID: DE42EE9B14F96465
4 changed files with 111 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -561,3 +561,55 @@ class TC_00_qvm_run(qubesadmin.tests.QubesTestCase):
('test-vm', 'qubes.VMShell', b'command& exit\n')
])
self.assertAllCalled()
def test_020_run_exec_with_vmexec_not_supported(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.feature.CheckWithTemplate', 'os', None)] = \
b'2\x00QubesFeatureNotFoundError\x00\x00Feature \'os\' not set\x00'
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.List', None, None)] = \
# b'0\x00test-vm class=AppVM state=Running\n'
ret = qubesadmin.tools.qvm_run.main(
['--no-gui', 'test-vm', 'command', 'arg'],
app=self.app)
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('test-vm', 'qubes.VMShell', {
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
}),
('test-vm', 'qubes.VMShell', b'command arg; exit\n')
])
self.assertAllCalled()
def test_020_run_exec_with_vmexec_supported(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.feature.CheckWithTemplate',
'vmexec', None)] = \
b'0\x001'
# self.app.expected_calls[
# ('test-vm', 'admin.vm.List', None, None)] = \
# b'0\x00test-vm class=AppVM state=Running\n'
ret = qubesadmin.tools.qvm_run.main(
['--no-gui', 'test-vm', 'command', 'arg'],
app=self.app)
self.assertEqual(ret, 0)
self.assertEqual(self.app.service_calls, [
('test-vm', 'qubes.VMExec+command+arg', {
'stdout': subprocess.DEVNULL,
'stderr': subprocess.DEVNULL,
'user': None,
}),
('test-vm', 'qubes.VMExec+command+arg', b'')
])
self.assertAllCalled()

11
qubesadmin/tests/utils.py
View File

@ -110,3 +110,14 @@ class TestVMUsage(qubesadmin.tests.QubesTestCase):
self.app.domains['sys-net'])
self.assertListEqual(result, [(self.app.domains['vm1'], 'netvm')])
class TestVMExecEncode(qubesadmin.tests.QubesTestCase):
def test_00_encode(self):
self.assertEqual(
qubesadmin.utils.encode_for_vmexec(['ls', '-a']),
'ls+--a')
self.assertEqual(
qubesadmin.utils.encode_for_vmexec(
['touch', '/home/user/.profile']),
'touch+-2Fhome-2Fuser-2F.profile')

33
qubesadmin/tools/qvm_run.py
View File

@ -1,4 +1,4 @@
# -*- encoding: utf8 -*-
# -*- encoding: utf-8 -*-
#
# The Qubes OS Project, http://www.qubes-os.org
#
@ -21,6 +21,7 @@
''' qvm-run tool'''
import contextlib
import os
import shlex
import signal
import subprocess
import sys
@ -31,6 +32,7 @@ import select
import qubesadmin.tools
import qubesadmin.exc
import qubesadmin.utils
parser = qubesadmin.tools.QubesArgumentParser()
@ -90,6 +92,9 @@ parser.add_argument('--service',
action='store_true', dest='service',
help='run a qrexec service (named by COMMAND) instead of shell command')
parser.add_argument('--no-shell', action='store_true',
help='treat COMMAND as a simple executable, not a shell command')
target_parser = parser.add_mutually_exclusive_group()
target_parser.add_argument('--dispvm', action='store', nargs='?',
@ -111,6 +116,9 @@ parser.add_argument('--exclude', action='append', default=[],
parser.add_argument('cmd', metavar='COMMAND',
help='command or service to run')
parser.add_argument('cmd_args', nargs='*', metavar='ARG',
help='command arguments (implies --no-shell)')
def copy_stdin(stream):
'''Copy stdin to *stream*'''
# multiprocessing.Process have sys.stdin connected to /dev/null, use fd 0
@ -164,19 +172,36 @@ def run_command_single(args, vm):
# simultaneous vchan connections
run_kwargs['wait'] = False
use_exec = len(args.cmd_args) > 0 or args.no_shell
copy_proc = None
local_proc = None
shell_cmd = None
if args.service:
service = args.cmd
elif use_exec:
all_args = [args.cmd] + args.cmd_args
if vm.features.check_with_template('vmexec', False):
service = 'qubes.VMExec'
if args.gui and args.dispvm:
service = 'qubes.VMExecGUI'
service += '+' + qubesadmin.utils.encode_for_vmexec(all_args)
else:
service = 'qubes.VMShell'
if args.gui and args.dispvm:
service += '+WaitForSession'
shell_cmd = ' '.join(shlex.quote(arg) for arg in all_args)
else:
service = 'qubes.VMShell'
if args.gui and args.dispvm:
service += '+WaitForSession'
shell_cmd = args.cmd
proc = vm.run_service(service,
user=args.user,
**run_kwargs)
if not args.service:
proc.stdin.write(vm.prepare_input_for_vmshell(args.cmd))
if shell_cmd:
proc.stdin.write(vm.prepare_input_for_vmshell(shell_cmd))
proc.stdin.flush()
if args.localcmd:
local_proc = subprocess.Popen(args.localcmd,
@ -211,6 +236,8 @@ def main(args=None, app=None):
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')
if args.service and args.no_shell:
parser.error('--no-shell does not apply to --service')
retcode = 0

18
qubesadmin/utils.py
View File

@ -24,6 +24,7 @@
"""Various utility functions."""
import os
import re
import qubesadmin.exc
@ -146,3 +147,20 @@ def vm_dependencies(app, reference_vm):
result.append((vm, prop))
return result
def encode_for_vmexec(args):
"""
Encode an argument list for qubes.VMExec call.
"""
def encode(part):
if part.group(0) == b'-':
return b'--'
return '-{:02X}'.format(ord(part.group(0))).encode('ascii')
parts = []
for arg in args:
part = re.sub(br'[^a-zA-Z0-9_.+]', encode, arg.encode('utf-8'))
parts.append(part)
return b'+'.join(parts).decode('ascii')