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

Add support for run_service(..., filter_esc=True) in a VM

Browse Source 
Since qrexec-client-vm got support for filtering escape characters, use
it here too.

QubesOS/qubes-issues#5322
This commit is contained in:
Marek Marczykowski-Górecki 2019-09-21 03:47:30 +02:00
parent d7dcdd3a0b
commit 1fcb031192
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
2 changed files with 22 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
qubesadmin/app.py
View File

@ -678,14 +678,16 @@ class QubesRemote(QubesBase):
:param bool wait: wait for process to finish
:rtype: subprocess.Popen
"""
if filter_esc:
raise NotImplementedError(
'filter_esc not implemented for calls from VM')
if user:
raise ValueError(
'non-default user not possible for calls from VM')
if not wait and localcmd:
raise ValueError('wait=False incompatible with localcmd')
qrexec_opts = []
if filter_esc:
qrexec_opts.extend(['-t'])
if filter_esc or os.isatty(sys.stderr.fileno()):
qrexec_opts.extend(['-T'])
if not wait:
# qrexec-client-vm can only request service calls, which are
# started using MSG_EXEC_CMDLINE qrexec protocol message; this
@ -700,6 +702,9 @@ class QubesRemote(QubesBase):
kwargs.setdefault('stdout', subprocess.PIPE)
kwargs.setdefault('stderr', subprocess.PIPE)
proc = subprocess.Popen(
[qubesadmin.config.QREXEC_CLIENT_VM, dest or '', service] + (
shlex.split(localcmd) if localcmd else []), **kwargs)
[qubesadmin.config.QREXEC_CLIENT_VM] +
qrexec_opts +
[dest or '', service] +
(shlex.split(localcmd) if localcmd else []),
**kwargs)
return proc

16
qubesadmin/tests/app.py
View File

@ -880,26 +880,34 @@ class TC_30_QubesRemote(unittest.TestCase):
])
self.assertEqual(value, b'return-value')
@mock.patch('os.isatty', lambda fd: fd == 2)
def test_010_run_service(self):
self.app.run_service('some-vm', 'service.name')
self.proc_mock.assert_called_once_with([
qubesadmin.config.QREXEC_CLIENT_VM,
'some-vm', 'service.name'],
'-T', 'some-vm', 'service.name'],
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
@mock.patch('os.isatty', lambda fd: fd == 2)
def test_011_run_service_filter_esc(self):
with self.assertRaises(NotImplementedError):
p = self.app.run_service('some-vm', 'service.name', filter_esc=True)
self.app.run_service('some-vm', 'service.name', filter_esc=True)
self.proc_mock.assert_called_once_with([
qubesadmin.config.QREXEC_CLIENT_VM,
'-t', '-T', 'some-vm', 'service.name'],
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
@mock.patch('os.isatty', lambda fd: fd == 2)
def test_012_run_service_user(self):
with self.assertRaises(ValueError):
p = self.app.run_service('some-vm', 'service.name', user='user')
@mock.patch('os.isatty', lambda fd: fd == 2)
def test_013_run_service_default_target(self):
self.app.run_service('', 'service.name')
self.proc_mock.assert_called_once_with([
qubesadmin.config.QREXEC_CLIENT_VM,
'', 'service.name'],
'-T', '', 'service.name'],
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE)