diff --git a/devlib/host.py b/devlib/host.py
index 70461fa..48e785f 100644
--- a/devlib/host.py
+++ b/devlib/host.py
@@ -20,6 +20,7 @@ import subprocess
import logging
from distutils.dir_util import copy_tree
from getpass import getpass
+from pipes import quote
from devlib.exception import TargetTransientError, TargetStableError
from devlib.utils.misc import check_output
@@ -70,7 +71,7 @@ class LocalConnection(object):
if self.unrooted:
raise TargetStableError('unrooted')
password = self._get_password()
- command = 'echo \'{}\' | sudo -S '.format(password) + command
+ command = 'echo {} | sudo -S '.format(quote(password)) + command
ignore = None if check_exit_code else 'all'
try:
return check_output(command, shell=True, timeout=timeout, ignore=ignore)[0]
@@ -87,7 +88,7 @@ class LocalConnection(object):
if self.unrooted:
raise TargetStableError('unrooted')
password = self._get_password()
- command = 'echo \'{}\' | sudo -S '.format(password) + command
+ command = 'echo {} | sudo -S '.format(quote(password)) + command
return subprocess.Popen(command, stdout=stdout, stderr=stderr, shell=True)
def close(self):
diff --git a/devlib/instrument/acmecape.py b/devlib/instrument/acmecape.py
index 16a3e1b..329c9ee 100644
--- a/devlib/instrument/acmecape.py
+++ b/devlib/instrument/acmecape.py
@@ -22,6 +22,7 @@ import tempfile
from fcntl import fcntl, F_GETFL, F_SETFL
from string import Template
from subprocess import Popen, PIPE, STDOUT
+from pipes import quote
from devlib import Instrument, CONTINUOUS, MeasurementsCsv
from devlib.exception import HostError
@@ -89,6 +90,7 @@ class AcmeCapeInstrument(Instrument):
iio_device=self.iio_device,
outfile=self.raw_data_file
)
+ params = {k: quote(v) for k, v in params.items()}
self.command = IIOCAP_CMD_TEMPLATE.substitute(**params)
self.logger.debug('ACME cape command: {}'.format(self.command))
diff --git a/devlib/instrument/arm_energy_probe.py b/devlib/instrument/arm_energy_probe.py
index 05ca6d6..4db97d8 100644
--- a/devlib/instrument/arm_energy_probe.py
+++ b/devlib/instrument/arm_energy_probe.py
@@ -34,6 +34,7 @@ from __future__ import division
import os
import subprocess
import signal
+from pipes import quote
import tempfile
import shutil
@@ -97,7 +98,7 @@ class ArmEnergyProbeInstrument(Instrument):
self.output_file_figure = os.path.join(self.output_directory, 'summary.txt')
self.output_file_error = os.path.join(self.output_directory, 'error.log')
self.output_fd_error = open(self.output_file_error, 'w')
- self.command = 'arm-probe --config {} > {}'.format(self.config_file, self.output_file_raw)
+ self.command = 'arm-probe --config {} > {}'.format(quote(self.config_file), quote(self.output_file_raw))
def start(self):
self.logger.debug(self.command)
diff --git a/devlib/instrument/energy_probe.py b/devlib/instrument/energy_probe.py
index 16aa7e5..4175a5d 100644
--- a/devlib/instrument/energy_probe.py
+++ b/devlib/instrument/energy_probe.py
@@ -19,6 +19,7 @@ import tempfile
import struct
import subprocess
import sys
+from pipes import quote
from devlib.instrument import Instrument, CONTINUOUS, MeasurementsCsv
from devlib.exception import HostError
@@ -65,7 +66,10 @@ class EnergyProbeInstrument(Instrument):
parts = ['-r {}:{} '.format(i, int(1000 * rval))
for i, rval in enumerate(self.resistor_values)]
rstring = ''.join(parts)
- self.command = '{} -d {} -l {} {}'.format(self.caiman, self.device_entry, rstring, self.raw_output_directory)
+ self.command = '{} -d {} -l {} {}'.format(
+ quote(self.caiman), quote(self.device_entry),
+ rstring, quote(self.raw_output_directory)
+ )
self.raw_data_file = None
def start(self):
diff --git a/devlib/platform/gem5.py b/devlib/platform/gem5.py
index 817699a..f197f97 100644
--- a/devlib/platform/gem5.py
+++ b/devlib/platform/gem5.py
@@ -18,6 +18,7 @@ import subprocess
import shutil
import time
import types
+from pipes import quote
from devlib.exception import TargetStableError
from devlib.host import PACKAGE_BIN_DIRECTORY
@@ -129,7 +130,7 @@ class Gem5SimulationPlatform(Platform):
self.logger.info("Starting the gem5 simulator")
command_line = "{} --outdir={} {} {}".format(self.gem5args_binary,
- self.gem5_out_dir,
+ quote(self.gem5_out_dir),
self.gem5args_args,
self.gem5args_virtio)
self.logger.debug("gem5 command line: {}".format(command_line))
diff --git a/devlib/target.py b/devlib/target.py
index a221eba..12d8b7d 100644
--- a/devlib/target.py
+++ b/devlib/target.py
@@ -118,7 +118,7 @@ class Target(object):
@property
@memoized
def kernel_version(self):
- return KernelVersion(self.execute('{} uname -r -v'.format(self.busybox)).strip())
+ return KernelVersion(self.execute('{} uname -r -v'.format(quote(self.busybox))).strip())
@property
def os_version(self): # pylint: disable=no-self-use
@@ -156,7 +156,7 @@ class Target(object):
except TargetStableError:
for path in ['/boot/config', '/boot/config-$(uname -r)']:
try:
- return KernelConfig(self.execute('cat {}'.format(path)))
+ return KernelConfig(self.execute('cat {}'.format(quote(path))))
except TargetStableError:
pass
return KernelConfig('')
@@ -253,8 +253,8 @@ class Target(object):
if check_boot_completed:
self.wait_boot_complete(timeout)
self._resolve_paths()
- self.execute('mkdir -p {}'.format(self.working_directory))
- self.execute('mkdir -p {}'.format(self.executables_directory))
+ self.execute('mkdir -p {}'.format(quote(self.working_directory)))
+ self.execute('mkdir -p {}'.format(quote(self.executables_directory)))
self.busybox = self.install(os.path.join(PACKAGE_BIN_DIRECTORY, self.abi, 'busybox'))
self.platform.update_from_target(self)
self._update_modules('connected')
@@ -286,7 +286,7 @@ class Target(object):
# Initialize modules which requires Buxybox (e.g. shutil dependent tasks)
self._update_modules('setup')
- self.execute('mkdir -p {}'.format(self._file_transfer_cache))
+ self.execute('mkdir -p {}'.format(quote(self._file_transfer_cache)))
def reboot(self, hard=False, connect=True, timeout=180):
if hard:
@@ -320,20 +320,20 @@ class Target(object):
self.conn.push(source, dest, timeout=timeout)
else:
device_tempfile = self.path.join(self._file_transfer_cache, source.lstrip(self.path.sep))
- self.execute("mkdir -p '{}'".format(self.path.dirname(device_tempfile)))
+ self.execute("mkdir -p {}".format(quote(self.path.dirname(device_tempfile))))
self.conn.push(source, device_tempfile, timeout=timeout)
- self.execute("cp '{}' '{}'".format(device_tempfile, dest), as_root=True)
+ self.execute("cp {} {}".format(quote(device_tempfile), quote(dest)), as_root=True)
def pull(self, source, dest, as_root=False, timeout=None): # pylint: disable=arguments-differ
if not as_root:
self.conn.pull(source, dest, timeout=timeout)
else:
device_tempfile = self.path.join(self._file_transfer_cache, source.lstrip(self.path.sep))
- self.execute("mkdir -p '{}'".format(self.path.dirname(device_tempfile)))
- self.execute("cp -r '{}' '{}'".format(source, device_tempfile), as_root=True)
- self.execute("chmod 0644 '{}'".format(device_tempfile), as_root=True)
+ self.execute("mkdir -p {}".format(quote(self.path.dirname(device_tempfile))))
+ self.execute("cp -r {} {}".format(quote(source), quote(device_tempfile)), as_root=True)
+ self.execute("chmod 0644 {}".format(quote(device_tempfile)), as_root=True)
self.conn.pull(device_tempfile, dest, timeout=timeout)
- self.execute("rm -r '{}'".format(device_tempfile), as_root=True)
+ self.execute("rm -r {}".format(quote(device_tempfile)), as_root=True)
def get_directory(self, source_dir, dest, as_root=False):
""" Pull a directory from the device, after compressing dir """
@@ -350,11 +350,12 @@ class Target(object):
tar_file_name = self.path.join(self._file_transfer_cache, tar_file_name)
# Does the folder exist?
- self.execute('ls -la {}'.format(source_dir), as_root=as_root)
+ self.execute('ls -la {}'.format(quote(source_dir)), as_root=as_root)
# Try compressing the folder
try:
- self.execute('{} tar -cvf {} {}'.format(self.busybox, tar_file_name,
- source_dir), as_root=as_root)
+ self.execute('{} tar -cvf {} {}'.format(
+ quote(self.busybox), quote(tar_file_name), quote(source_dir)
+ ), as_root=as_root)
except TargetStableError:
self.logger.debug('Failed to run tar command on target! ' \
'Not pulling directory {}'.format(source_dir))
@@ -407,9 +408,9 @@ class Target(object):
command = '{} {}'.format(command, args)
if on_cpus:
on_cpus = bitmask(on_cpus)
- command = '{} taskset 0x{:x} {}'.format(self.busybox, on_cpus, command)
+ command = '{} taskset 0x{:x} {}'.format(quote(self.busybox), on_cpus, command)
if in_directory:
- command = 'cd {} && {}'.format(in_directory, command)
+ command = 'cd {} && {}'.format(quote(in_directory), command)
if redirect_stderr:
command = '{} 2>&1'.format(command)
return self.execute(command, as_root=as_root, timeout=timeout)
@@ -441,9 +442,9 @@ class Target(object):
command = '{} {}'.format(command, args)
if on_cpus:
on_cpus = bitmask(on_cpus)
- command = '{} taskset 0x{:x} {}'.format(self.busybox, on_cpus, command)
+ command = '{} taskset 0x{:x} {}'.format(quote(self.busybox), on_cpus, command)
if in_directory:
- command = 'cd {} && {}'.format(in_directory, command)
+ command = 'cd {} && {}'.format(quote(in_directory), command)
return self.background(command, as_root=as_root)
def kick_off(self, command, as_root=False):
@@ -452,7 +453,7 @@ class Target(object):
# sysfs interaction
def read_value(self, path, kind=None):
- output = self.execute('cat \'{}\''.format(path), as_root=self.needs_su).strip() # pylint: disable=E1103
+ output = self.execute('cat {}'.format(quote(path)), as_root=self.needs_su).strip() # pylint: disable=E1103
if kind:
return kind(output)
else:
@@ -466,7 +467,7 @@ class Target(object):
def write_value(self, path, value, verify=True):
value = str(value)
- self.execute('echo {} > \'{}\''.format(value, path), check_exit_code=False, as_root=True)
+ self.execute('echo {} > {}'.format(quote(value), quote(path)), check_exit_code=False, as_root=True)
if verify:
output = self.read_value(path)
if not output == value:
@@ -512,12 +513,12 @@ class Target(object):
# files
def file_exists(self, filepath):
- command = 'if [ -e \'{}\' ]; then echo 1; else echo 0; fi'
- output = self.execute(command.format(filepath), as_root=self.is_rooted)
+ command = 'if [ -e {} ]; then echo 1; else echo 0; fi'
+ output = self.execute(command.format(quote(filepath)), as_root=self.is_rooted)
return boolean(output.strip())
def directory_exists(self, filepath):
- output = self.execute('if [ -d \'{}\' ]; then echo 1; else echo 0; fi'.format(filepath))
+ output = self.execute('if [ -d {} ]; then echo 1; else echo 0; fi'.format(quote(filepath)))
# output from ssh my contain part of the expression in the buffer,
# split out everything except the last word.
return boolean(output.split()[-1]) # pylint: disable=maybe-no-member
@@ -642,7 +643,7 @@ class Target(object):
def insmod(self, path):
target_path = self.get_workpath(os.path.basename(path))
self.push(path, target_path)
- self.execute('insmod {}'.format(target_path), as_root=True)
+ self.execute('insmod {}'.format(quote(target_path)), as_root=True)
def extract(self, path, dest=None):
@@ -683,7 +684,7 @@ class Target(object):
self.execute('sleep {}'.format(duration), timeout=timeout)
def read_tree_values_flat(self, path, depth=1, check_exit_code=True):
- command = 'read_tree_values {} {}'.format(path, depth)
+ command = 'read_tree_values {} {}'.format(quote(path), depth)
output = self._execute_util(command, as_root=self.is_rooted,
check_exit_code=check_exit_code)
@@ -731,17 +732,17 @@ class Target(object):
extracted = dest
else:
extracted = self.path.dirname(path)
- cmdtext = cmd.format(self.busybox, path, extracted)
+ cmdtext = cmd.format(quote(self.busybox), quote(path), quote(extracted))
self.execute(cmdtext)
return extracted
def _extract_file(self, path, cmd, dest=None):
cmd = '{} ' + cmd # busybox
- cmdtext = cmd.format(self.busybox, path)
+ cmdtext = cmd.format(quote(self.busybox), quote(path))
self.execute(cmdtext)
extracted = self.path.splitext(path)[0]
if dest:
- self.execute('mv -f {} {}'.format(extracted, dest))
+ self.execute('mv -f {} {}'.format(quote(extracted), quote(dest)))
if dest.endswith('/'):
extracted = self.path.join(dest, self.path.basename(extracted))
else:
@@ -785,7 +786,7 @@ class Target(object):
# It would be nice to use busybox for this, but that means we'd need
# root (ping is usually setuid so it can open raw sockets to send ICMP)
command = 'ping -q -c 1 -w {} {} 2>&1'.format(timeout_s,
- GOOGLE_DNS_SERVER_ADDRESS)
+ quote(GOOGLE_DNS_SERVER_ADDRESS))
# We'll use our own retrying mechanism (rather than just using ping's -c
# to send multiple packets) so that we don't slow things down in the
@@ -896,7 +897,7 @@ class LinuxTarget(Target):
def get_pids_of(self, process_name):
"""Returns a list of PIDs of all processes with the specified name."""
# result should be a column of PIDs with the first row as "PID" header
- result = self.execute('ps -C {} -o pid'.format(process_name), # NOQA
+ result = self.execute('ps -C {} -o pid'.format(quote(process_name)), # NOQA
check_exit_code=False).strip().split()
if len(result) >= 2: # at least one row besides the header
return list(map(int, result[1:]))
@@ -931,7 +932,7 @@ class LinuxTarget(Target):
destpath = self.path.join(self.executables_directory,
with_name and with_name or self.path.basename(filepath))
self.push(filepath, destpath)
- self.execute('chmod a+x {}'.format(destpath), timeout=timeout)
+ self.execute('chmod a+x {}'.format(quote(destpath)), timeout=timeout)
self._installed_binaries[self.path.basename(destpath)] = destpath
return destpath
@@ -947,7 +948,7 @@ class LinuxTarget(Target):
tmpfile = self.tempfile()
cmd = 'DISPLAY=:0.0 scrot {} && {} date -u -Iseconds'
- ts = self.execute(cmd.format(tmpfile, self.busybox)).strip()
+ ts = self.execute(cmd.format(quote(tmpfile), quote(self.busybox))).strip()
filepath = filepath.format(ts=ts)
self.pull(tmpfile, filepath)
self.remove(tmpfile)
@@ -1125,7 +1126,7 @@ class AndroidTarget(Target):
if as_root is None:
as_root = self.needs_su
try:
- command = 'cd {} && {} nohup {} &'.format(self.working_directory, self.busybox, command)
+ command = 'cd {} && {} nohup {} &'.format(quote(self.working_directory), quote(self.busybox), command)
self.execute(command, timeout=1, as_root=as_root)
except TimeoutError:
pass
@@ -1139,14 +1140,14 @@ class AndroidTarget(Target):
# so we try the new version, and if it fails we use the old version.
self.ls_command = 'ls -1'
try:
- self.execute('ls -1 {}'.format(self.working_directory), as_root=False)
+ self.execute('ls -1 {}'.format(quote(self.working_directory)), as_root=False)
except TargetStableError:
self.ls_command = 'ls'
def list_directory(self, path, as_root=False):
if self.ls_command == '':
self.__setup_list_directory()
- contents = self.execute('{} "{}"'.format(self.ls_command, escape_double_quotes(path)), as_root=as_root)
+ contents = self.execute('{} {}'.format(self.ls_command, quote(path)), as_root=as_root)
return [x.strip() for x in contents.split('\n') if x.strip()]
def install(self, filepath, timeout=None, with_name=None): # pylint: disable=W0221
@@ -1194,7 +1195,7 @@ class AndroidTarget(Target):
def capture_screen(self, filepath):
on_device_file = self.path.join(self.working_directory, 'screen_capture.png')
cmd = 'screencap -p {} && {} date -u -Iseconds'
- ts = self.execute(cmd.format(on_device_file, self.busybox)).strip()
+ ts = self.execute(cmd.format(quote(on_device_file), quote(self.busybox))).strip()
filepath = filepath.format(ts=ts)
self.pull(on_device_file, filepath)
self.remove(on_device_file)
@@ -1285,14 +1286,14 @@ class AndroidTarget(Target):
return output.split()
def get_package_version(self, package):
- output = self.execute('dumpsys package {}'.format(package))
+ output = self.execute('dumpsys package {}'.format(quote(package)))
for line in convert_new_lines(output).split('\n'):
if 'versionName' in line:
return line.split('=', 1)[1]
return None
def get_package_info(self, package):
- output = self.execute('pm list packages -f {}'.format(package))
+ output = self.execute('pm list packages -f {}'.format(quote(package)))
for entry in output.strip().split('\n'):
rest, entry_package = entry.rsplit('=', 1)
if entry_package != package:
@@ -1317,13 +1318,13 @@ class AndroidTarget(Target):
if self.get_sdk_version() >= 23:
flags.append('-g') # Grant all runtime permissions
self.logger.debug("Replace APK = {}, ADB flags = '{}'".format(replace, ' '.join(flags)))
- return adb_command(self.adb_name, "install {} '{}'".format(' '.join(flags), filepath), timeout=timeout)
+ return adb_command(self.adb_name, "install {} {}".format(' '.join(flags), quote(filepath)), timeout=timeout)
else:
raise TargetStableError('Can\'t install {}: unsupported format.'.format(filepath))
def grant_package_permission(self, package, permission):
try:
- return self.execute('pm grant {} {}'.format(package, permission))
+ return self.execute('pm grant {} {}'.format(quote(package), quote(permission)))
except TargetStableError as e:
if 'is not a changeable permission type' in e.message:
pass # Ignore if unchangeable
@@ -1353,16 +1354,16 @@ class AndroidTarget(Target):
"""
Force a re-index of the mediaserver cache for the specified file.
"""
- command = 'am broadcast -a android.intent.action.MEDIA_SCANNER_SCAN_FILE -d file://'
- self.execute(command + filepath)
+ command = 'am broadcast -a android.intent.action.MEDIA_SCANNER_SCAN_FILE -d {}'
+ self.execute(command.format(quote('file://' + filepath)))
def broadcast_media_mounted(self, dirpath, as_root=False):
"""
Force a re-index of the mediaserver cache for the specified directory.
"""
- command = 'am broadcast -a android.intent.action.MEDIA_MOUNTED -d file://{} '\
+ command = 'am broadcast -a android.intent.action.MEDIA_MOUNTED -d {} '\
'-n com.android.providers.media/.MediaScannerReceiver'
- self.execute(command.format(dirpath), as_root=as_root)
+ self.execute(command.format(quote('file://'+dirpath)), as_root=as_root)
def install_executable(self, filepath, with_name=None):
self._ensure_executables_directory_is_writable()
@@ -1371,14 +1372,14 @@ class AndroidTarget(Target):
on_device_executable = self.path.join(self.executables_directory, executable_name)
self.push(filepath, on_device_file)
if on_device_file != on_device_executable:
- self.execute('cp {} {}'.format(on_device_file, on_device_executable), as_root=self.needs_su)
+ self.execute('cp {} {}'.format(quote(on_device_file), quote(on_device_executable)), as_root=self.needs_su)
self.remove(on_device_file, as_root=self.needs_su)
- self.execute("chmod 0777 '{}'".format(on_device_executable), as_root=self.needs_su)
+ self.execute("chmod 0777 {}".format(quote(on_device_executable)), as_root=self.needs_su)
self._installed_binaries[executable_name] = on_device_executable
return on_device_executable
def uninstall_package(self, package):
- adb_command(self.adb_name, "uninstall {}".format(package), timeout=30)
+ adb_command(self.adb_name, "uninstall {}".format(quote(package)), timeout=30)
def uninstall_executable(self, executable_name):
on_device_executable = self.path.join(self.executables_directory, executable_name)
@@ -1387,8 +1388,8 @@ class AndroidTarget(Target):
def dump_logcat(self, filepath, filter=None, append=False, timeout=30): # pylint: disable=redefined-builtin
op = '>>' if append else '>'
- filtstr = ' -s {}'.format(filter) if filter else ''
- command = 'logcat -d{} {} {}'.format(filtstr, op, filepath)
+ filtstr = ' -s {}'.format(quote(filter)) if filter else ''
+ command = 'logcat -d{} {} {}'.format(filtstr, op, quote(filepath))
adb_command(self.adb_name, command, timeout=timeout)
def clear_logcat(self):
@@ -1547,8 +1548,8 @@ class AndroidTarget(Target):
if matched:
entry = sorted(matched, key=lambda x: len(x.mount_point))[-1]
if 'rw' not in entry.options:
- self.execute('mount -o rw,remount {} {}'.format(entry.device,
- entry.mount_point),
+ self.execute('mount -o rw,remount {} {}'.format(quote(entry.device),
+ quote(entry.mount_point)),
as_root=True)
else:
message = 'Could not find mount point for executables directory {}'
diff --git a/devlib/utils/android.py b/devlib/utils/android.py
index ef3cb01..09f4e92 100755
--- a/devlib/utils/android.py
+++ b/devlib/utils/android.py
@@ -235,7 +235,7 @@ class AdbConnection(object):
def push(self, source, dest, timeout=None):
if timeout is None:
timeout = self.timeout
- command = "push '{}' '{}'".format(source, dest)
+ command = "push {} {}".format(quote(source), quote(dest))
if not os.path.exists(source):
raise HostError('No such file "{}"'.format(source))
return adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server)
@@ -249,10 +249,10 @@ class AdbConnection(object):
command = 'shell {} {}'.format(self.ls_command, source)
output = adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server)
for line in output.splitlines():
- command = "pull '{}' '{}'".format(line.strip(), dest)
+ command = "pull {} {}".format(quote(line.strip()), quote(dest))
adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server)
return
- command = "pull '{}' '{}'".format(source, dest)
+ command = "pull {} {}".format(quote(source), quote(dest))
return adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server)
# pylint: disable=unused-argument
@@ -285,7 +285,7 @@ class AdbConnection(object):
def fastboot_command(command, timeout=None, device=None):
_check_env()
- target = '-s {}'.format(device) if device else ''
+ target = '-s {}'.format(quote(device)) if device else ''
full_command = 'fastboot {} {}'.format(target, command)
logger.debug(full_command)
output, _ = check_output(full_command, timeout, shell=True)
@@ -293,7 +293,7 @@ def fastboot_command(command, timeout=None, device=None):
def fastboot_flash_partition(partition, path_to_image):
- command = 'flash {} {}'.format(partition, path_to_image)
+ command = 'flash {} {}'.format(quote(partition), quote(path_to_image))
fastboot_command(command)
@@ -341,7 +341,7 @@ def adb_connect(device, timeout=None, attempts=MAX_ATTEMPTS):
tries += 1
if device:
if "." in device: # Connect is required only for ADB-over-IP
- command = 'adb connect {}'.format(device)
+ command = 'adb connect {}'.format(quote(device))
logger.debug(command)
output, _ = check_output(command, shell=True, timeout=timeout)
if _ping(device):
@@ -368,7 +368,7 @@ def adb_disconnect(device):
def _ping(device):
_check_env()
- device_string = ' -s {}'.format(device) if device else ''
+ device_string = ' -s {}'.format(quote(device)) if device else ''
command = "adb{} shell \"ls /data/local/tmp > /dev/null\"".format(device_string)
logger.debug(command)
result = subprocess.call(command, stderr=subprocess.PIPE, shell=True)
@@ -610,9 +610,9 @@ class LogcatMonitor(object):
# Logcat on older version of android do not support the -e argument
# so fall back to using grep.
if self.target.get_sdk_version() > 23:
- logcat_cmd = '{} -e "{}"'.format(logcat_cmd, regexp)
+ logcat_cmd = '{} -e {}'.format(logcat_cmd, quote(regexp))
else:
- logcat_cmd = '{} | grep "{}"'.format(logcat_cmd, regexp)
+ logcat_cmd = '{} | grep {}'.format(logcat_cmd, quote(regexp))
logcat_cmd = get_adb_command(self.target.conn.device, logcat_cmd)
diff --git a/devlib/utils/misc.py b/devlib/utils/misc.py
index c7ec8f1..ce6b029 100644
--- a/devlib/utils/misc.py
+++ b/devlib/utils/misc.py
@@ -35,6 +35,7 @@ import subprocess
import sys
import threading
import wrapt
+import warnings
from past.builtins import basestring
@@ -416,6 +417,16 @@ def convert_new_lines(text):
""" Convert new lines to a common format. """
return text.replace('\r\n', '\n').replace('\r', '\n')
+def sanitize_cmd_template(cmd):
+ msg = (
+ '''Quoted placeholder should not be used, as it will result in quoting the text twice. {} should be used instead of '{}' or "{}" in the template: '''
+ )
+ for unwanted in ('"{}"', "'{}'"):
+ if unwanted in cmd:
+ warnings.warn(msg + cmd, stacklevel=2)
+ cmd = cmd.replace(unwanted, '{}')
+
+ return cmd
def escape_quotes(text):
"""Escape quotes, and escaped quotes, in the specified text."""
diff --git a/devlib/utils/rendering.py b/devlib/utils/rendering.py
index fa30b3f..a72a72e 100644
--- a/devlib/utils/rendering.py
+++ b/devlib/utils/rendering.py
@@ -21,6 +21,7 @@ import tempfile
import threading
import time
from collections import namedtuple
+from pipes import quote
# pylint: disable=redefined-builtin
from devlib.exception import WorkerThreadError, TargetNotRespondingError, TimeoutError
@@ -138,8 +139,8 @@ class SurfaceFlingerFrameCollector(FrameCollector):
self.target.execute('dumpsys SurfaceFlinger --latency-clear ')
def get_latencies(self, activity):
- cmd = 'dumpsys SurfaceFlinger --latency "{}"'
- return self.target.execute(cmd.format(activity))
+ cmd = 'dumpsys SurfaceFlinger --latency {}'
+ return self.target.execute(cmd.format(quote(activity)))
def list(self):
text = self.target.execute('dumpsys SurfaceFlinger --list')
diff --git a/devlib/utils/ssh.py b/devlib/utils/ssh.py
index bdd4345..8936606 100644
--- a/devlib/utils/ssh.py
+++ b/devlib/utils/ssh.py
@@ -26,6 +26,7 @@ import socket
import sys
import time
import atexit
+from pipes import quote
# pylint: disable=import-error,wrong-import-position,ungrouped-imports,wrong-import-order
import pexpect
@@ -39,7 +40,7 @@ from pexpect import EOF, TIMEOUT, spawn
# pylint: disable=redefined-builtin,wrong-import-position
from devlib.exception import (HostError, TargetStableError, TargetNotRespondingError,
TimeoutError, TargetTransientError)
-from devlib.utils.misc import which, strip_bash_colors, check_output
+from devlib.utils.misc import which, strip_bash_colors, check_output, sanitize_cmd_template
from devlib.utils.misc import (escape_single_quotes, escape_double_quotes,
escape_spaces)
from devlib.utils.types import boolean
@@ -169,7 +170,7 @@ class SshConnection(object):
password_prompt=None,
original_prompt=None,
platform=None,
- sudo_cmd="sudo -- sh -c '{}'"
+ sudo_cmd="sudo -- sh -c {}"
):
self.host = host
self.username = username
@@ -178,22 +179,18 @@ class SshConnection(object):
self.port = port
self.lock = threading.Lock()
self.password_prompt = password_prompt if password_prompt is not None else self.default_password_prompt
- self.sudo_cmd = sudo_cmd
+ self.sudo_cmd = sanitize_cmd_template(sudo_cmd)
logger.debug('Logging in {}@{}'.format(username, host))
timeout = timeout if timeout is not None else self.default_timeout
self.conn = ssh_get_shell(host, username, password, self.keyfile, port, timeout, False, None)
atexit.register(self.close)
def push(self, source, dest, timeout=30):
- dest = '"{}"@"{}":"{}"'.format(escape_double_quotes(self.username),
- escape_spaces(escape_double_quotes(self.host)),
- escape_spaces(escape_double_quotes(dest)))
+ dest = '{}@{}:{}'.format(self.username, self.host, dest)
return self._scp(source, dest, timeout)
def pull(self, source, dest, timeout=30):
- source = '"{}"@"{}":"{}"'.format(escape_double_quotes(self.username),
- escape_spaces(escape_double_quotes(self.host)),
- escape_spaces(escape_double_quotes(source)))
+ source = '{}@{}:{}'.format(self.username, self.host, source)
return self._scp(source, dest, timeout)
def execute(self, command, timeout=None, check_exit_code=True,
@@ -240,7 +237,7 @@ class SshConnection(object):
command = '{} {} {} {}@{} {}'.format(ssh, keyfile_string, port_string, self.username, self.host, command)
logger.debug(command)
if self.password:
- command = _give_password(self.password, command)
+ command, _ = _give_password(self.password, command)
return subprocess.Popen(command, stdout=stdout, stderr=stderr, shell=True)
except EOF:
raise TargetNotRespondingError('Connection lost.')
@@ -310,14 +307,13 @@ class SshConnection(object):
# fails to connect to a device if port is explicitly specified using -P
# option, even if it is the default port, 22. To minimize this problem,
# only specify -P for scp if the port is *not* the default.
- port_string = '-P {}'.format(self.port) if (self.port and self.port != 22) else ''
- keyfile_string = '-i {}'.format(self.keyfile) if self.keyfile else ''
- command = '{} -r {} {} {} {}'.format(scp, keyfile_string, port_string, source, dest)
+ port_string = '-P {}'.format(quote(str(self.port))) if (self.port and self.port != 22) else ''
+ keyfile_string = '-i {}'.format(quote(self.keyfile)) if self.keyfile else ''
+ command = '{} -r {} {} {} {}'.format(scp, keyfile_string, port_string, quote(source), quote(dest))
command_redacted = command
logger.debug(command)
if self.password:
- command = _give_password(self.password, command)
- command_redacted = command.replace(self.password, '<redacted>')
+ command, command_redacted = _give_password(self.password, command)
try:
check_output(command, timeout=timeout, shell=True)
except subprocess.CalledProcessError as e:
@@ -456,13 +452,12 @@ class Gem5Connection(TelnetConnection):
if os.path.basename(dest) != filename:
dest = os.path.join(dest, filename)
# Back to the gem5 world
- self._gem5_shell("ls -al {}{}".format(self.gem5_input_dir, filename))
- self._gem5_shell("cat '{}''{}' > '{}'".format(self.gem5_input_dir,
- filename,
- dest))
+ filename = quote(self.gem5_input_dir + filename)
+ self._gem5_shell("ls -al {}".format(filename))
+ self._gem5_shell("cat {} > {}".format(filename, quote(dest)))
self._gem5_shell("sync")
- self._gem5_shell("ls -al {}".format(dest))
- self._gem5_shell("ls -al {}".format(self.gem5_input_dir))
+ self._gem5_shell("ls -al {}".format(quote(dest)))
+ self._gem5_shell("ls -al {}".format(quote(self.gem5_input_dir)))
logger.debug("Push complete.")
def pull(self, source, dest, timeout=0): #pylint: disable=unused-argument
@@ -491,8 +486,8 @@ class Gem5Connection(TelnetConnection):
if os.path.isabs(source):
if os.path.dirname(source) != self.execute('pwd',
check_exit_code=False):
- self._gem5_shell("cat '{}' > '{}'".format(filename,
- dest_file))
+ self._gem5_shell("cat {} > {}".format(quote(filename),
+ quote(dest_file)))
self._gem5_shell("sync")
self._gem5_shell("ls -la {}".format(dest_file))
logger.debug('Finished the copy in the simulator')
@@ -875,7 +870,8 @@ class Gem5Connection(TelnetConnection):
"""
Internal method to check if the target has a certain file
"""
- command = 'if [ -e \'{}\' ]; then echo 1; else echo 0; fi'
+ filepath = quote(filepath)
+ command = 'if [ -e {} ]; then echo 1; else echo 0; fi'
output = self.execute(command.format(filepath), as_root=self.is_rooted)
return boolean(output.strip())
@@ -931,8 +927,10 @@ class AndroidGem5Connection(Gem5Connection):
def _give_password(password, command):
if not sshpass:
raise HostError('Must have sshpass installed on the host in order to use password-based auth.')
- pass_string = "sshpass -p '{}' ".format(password)
- return pass_string + command
+ pass_template = "sshpass -p {} "
+ pass_string = pass_template.format(quote(password))
+ redacted_string = pass_template.format(quote('<redacted>'))
+ return (pass_string + command, redacted_string + command)
def _check_env():