sascha/devlib
1
0
Fork 0
mirror of https://github.com/ARM-software/devlib.git synced 2025-02-07 13:40:48 +00:00
Code Issues Releases Wiki Activity

subprocess: Fix quoting issues

Browse Source 
Strings are passed to subprocess.Popen is used instead of sequences, so
proper quoting of parameters in command lines is needed to avoid
failures on filenames containing e.g. parenthesis. Use pipes.quote to
quote a string to avoid any issue with special characters.
This commit is contained in:
Douglas RAILLARD 2018-10-30 15:05:03 +00:00 committed by Marc Bonnici
parent d376bc10ee
commit ed7f0e56a2
10 changed files with 113 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
devlib/host.py
View File

@ -20,6 +20,7 @@ import subprocess
import logging import logging
from distutils.dir_util import copy_tree from distutils.dir_util import copy_tree
from getpass import getpass from getpass import getpass
from pipes import quote
from devlib.exception import TargetTransientError, TargetStableError from devlib.exception import TargetTransientError, TargetStableError
from devlib.utils.misc import check_output from devlib.utils.misc import check_output
@ -70,7 +71,7 @@ class LocalConnection(object):
if self.unrooted: if self.unrooted:
raise TargetStableError('unrooted') raise TargetStableError('unrooted')
password = self._get_password() 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' ignore = None if check_exit_code else 'all'
try: try:
return check_output(command, shell=True, timeout=timeout, ignore=ignore)[0] return check_output(command, shell=True, timeout=timeout, ignore=ignore)[0]
@ -87,7 +88,7 @@ class LocalConnection(object):
if self.unrooted: if self.unrooted:
raise TargetStableError('unrooted') raise TargetStableError('unrooted')
password = self._get_password() 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) return subprocess.Popen(command, stdout=stdout, stderr=stderr, shell=True)
def close(self): def close(self):

2
devlib/instrument/acmecape.py
View File

@ -22,6 +22,7 @@ import tempfile
from fcntl import fcntl, F_GETFL, F_SETFL from fcntl import fcntl, F_GETFL, F_SETFL
from string import Template from string import Template
from subprocess import Popen, PIPE, STDOUT from subprocess import Popen, PIPE, STDOUT
from pipes import quote
from devlib import Instrument, CONTINUOUS, MeasurementsCsv from devlib import Instrument, CONTINUOUS, MeasurementsCsv
from devlib.exception import HostError from devlib.exception import HostError
@ -89,6 +90,7 @@ class AcmeCapeInstrument(Instrument):
iio_device=self.iio_device, iio_device=self.iio_device,
outfile=self.raw_data_file outfile=self.raw_data_file
) )
params = {k: quote(v) for k, v in params.items()}
self.command = IIOCAP_CMD_TEMPLATE.substitute(**params) self.command = IIOCAP_CMD_TEMPLATE.substitute(**params)
self.logger.debug('ACME cape command: {}'.format(self.command)) self.logger.debug('ACME cape command: {}'.format(self.command))

3
devlib/instrument/arm_energy_probe.py
View File

@ -34,6 +34,7 @@ from __future__ import division
import os import os
import subprocess import subprocess
import signal import signal
from pipes import quote
import tempfile import tempfile
import shutil import shutil
@ -97,7 +98,7 @@ class ArmEnergyProbeInstrument(Instrument):
self.output_file_figure = os.path.join(self.output_directory, 'summary.txt') 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_file_error = os.path.join(self.output_directory, 'error.log')
self.output_fd_error = open(self.output_file_error, 'w') 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): def start(self):
self.logger.debug(self.command) self.logger.debug(self.command)

6
devlib/instrument/energy_probe.py
View File

@ -19,6 +19,7 @@ import tempfile
import struct import struct
import subprocess import subprocess
import sys import sys
from pipes import quote
from devlib.instrument import Instrument, CONTINUOUS, MeasurementsCsv from devlib.instrument import Instrument, CONTINUOUS, MeasurementsCsv
from devlib.exception import HostError from devlib.exception import HostError
@ -65,7 +66,10 @@ class EnergyProbeInstrument(Instrument):
parts = ['-r {}:{} '.format(i, int(1000 * rval)) parts = ['-r {}:{} '.format(i, int(1000 * rval))
for i, rval in enumerate(self.resistor_values)] for i, rval in enumerate(self.resistor_values)]
rstring = ''.join(parts) 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 self.raw_data_file = None
def start(self): def start(self):

3
devlib/platform/gem5.py
View File

@ -18,6 +18,7 @@ import subprocess
import shutil import shutil
import time import time
import types import types
from pipes import quote
from devlib.exception import TargetStableError from devlib.exception import TargetStableError
from devlib.host import PACKAGE_BIN_DIRECTORY from devlib.host import PACKAGE_BIN_DIRECTORY
@ -129,7 +130,7 @@ class Gem5SimulationPlatform(Platform):
self.logger.info("Starting the gem5 simulator") self.logger.info("Starting the gem5 simulator")
command_line = "{} --outdir={} {} {}".format(self.gem5args_binary, command_line = "{} --outdir={} {} {}".format(self.gem5args_binary,
self.gem5_out_dir, quote(self.gem5_out_dir),
self.gem5args_args, self.gem5args_args,
self.gem5args_virtio) self.gem5args_virtio)
self.logger.debug("gem5 command line: {}".format(command_line)) self.logger.debug("gem5 command line: {}".format(command_line))

103
devlib/target.py
View File

@ -118,7 +118,7 @@ class Target(object):
@property @property
@memoized @memoized
def kernel_version(self): 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 @property
def os_version(self): # pylint: disable=no-self-use def os_version(self): # pylint: disable=no-self-use
@ -156,7 +156,7 @@ class Target(object):
except TargetStableError: except TargetStableError:
for path in ['/boot/config', '/boot/config-$(uname -r)']: for path in ['/boot/config', '/boot/config-$(uname -r)']:
try: try:
return KernelConfig(self.execute('cat {}'.format(path))) return KernelConfig(self.execute('cat {}'.format(quote(path))))
except TargetStableError: except TargetStableError:
pass pass
return KernelConfig('') return KernelConfig('')
@ -253,8 +253,8 @@ class Target(object):
if check_boot_completed: if check_boot_completed:
self.wait_boot_complete(timeout) self.wait_boot_complete(timeout)
self._resolve_paths() self._resolve_paths()
self.execute('mkdir -p {}'.format(self.working_directory)) self.execute('mkdir -p {}'.format(quote(self.working_directory)))
self.execute('mkdir -p {}'.format(self.executables_directory)) self.execute('mkdir -p {}'.format(quote(self.executables_directory)))
self.busybox = self.install(os.path.join(PACKAGE_BIN_DIRECTORY, self.abi, 'busybox')) self.busybox = self.install(os.path.join(PACKAGE_BIN_DIRECTORY, self.abi, 'busybox'))
self.platform.update_from_target(self) self.platform.update_from_target(self)
self._update_modules('connected') self._update_modules('connected')
@ -286,7 +286,7 @@ class Target(object):
# Initialize modules which requires Buxybox (e.g. shutil dependent tasks) # Initialize modules which requires Buxybox (e.g. shutil dependent tasks)
self._update_modules('setup') 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): def reboot(self, hard=False, connect=True, timeout=180):
if hard: if hard:
@ -320,20 +320,20 @@ class Target(object):
self.conn.push(source, dest, timeout=timeout) self.conn.push(source, dest, timeout=timeout)
else: else:
device_tempfile = self.path.join(self._file_transfer_cache, source.lstrip(self.path.sep)) 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.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 def pull(self, source, dest, as_root=False, timeout=None): # pylint: disable=arguments-differ
if not as_root: if not as_root:
self.conn.pull(source, dest, timeout=timeout) self.conn.pull(source, dest, timeout=timeout)
else: else:
device_tempfile = self.path.join(self._file_transfer_cache, source.lstrip(self.path.sep)) 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.execute("cp -r '{}' '{}'".format(source, device_tempfile), as_root=True) self.execute("cp -r {} {}".format(quote(source), quote(device_tempfile)), as_root=True)
self.execute("chmod 0644 '{}'".format(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.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): def get_directory(self, source_dir, dest, as_root=False):
""" Pull a directory from the device, after compressing dir """ """ 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) tar_file_name = self.path.join(self._file_transfer_cache, tar_file_name)
# Does the folder exist? # 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 compressing the folder
try: try:
self.execute('{} tar -cvf {} {}'.format(self.busybox, tar_file_name, self.execute('{} tar -cvf {} {}'.format(
source_dir), as_root=as_root) quote(self.busybox), quote(tar_file_name), quote(source_dir)
), as_root=as_root)
except TargetStableError: except TargetStableError:
self.logger.debug('Failed to run tar command on target! ' \ self.logger.debug('Failed to run tar command on target! ' \
'Not pulling directory {}'.format(source_dir)) 'Not pulling directory {}'.format(source_dir))
@ -407,9 +408,9 @@ class Target(object):
command = '{} {}'.format(command, args) command = '{} {}'.format(command, args)
if on_cpus: if on_cpus:
on_cpus = bitmask(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: if in_directory:
command = 'cd {} && {}'.format(in_directory, command) command = 'cd {} && {}'.format(quote(in_directory), command)
if redirect_stderr: if redirect_stderr:
command = '{} 2>&1'.format(command) command = '{} 2>&1'.format(command)
return self.execute(command, as_root=as_root, timeout=timeout) return self.execute(command, as_root=as_root, timeout=timeout)
@ -441,9 +442,9 @@ class Target(object):
command = '{} {}'.format(command, args) command = '{} {}'.format(command, args)
if on_cpus: if on_cpus:
on_cpus = bitmask(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: if in_directory:
command = 'cd {} && {}'.format(in_directory, command) command = 'cd {} && {}'.format(quote(in_directory), command)
return self.background(command, as_root=as_root) return self.background(command, as_root=as_root)
def kick_off(self, command, as_root=False): def kick_off(self, command, as_root=False):
@ -452,7 +453,7 @@ class Target(object):
# sysfs interaction # sysfs interaction
def read_value(self, path, kind=None): 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: if kind:
return kind(output) return kind(output)
else: else:
@ -466,7 +467,7 @@ class Target(object):
def write_value(self, path, value, verify=True): def write_value(self, path, value, verify=True):
value = str(value) 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: if verify:
output = self.read_value(path) output = self.read_value(path)
if not output == value: if not output == value:
@ -512,12 +513,12 @@ class Target(object):
# files # files
def file_exists(self, filepath): def file_exists(self, filepath):
command = 'if [ -e \'{}\' ]; then echo 1; else echo 0; fi' command = 'if [ -e {} ]; then echo 1; else echo 0; fi'
output = self.execute(command.format(filepath), as_root=self.is_rooted) output = self.execute(command.format(quote(filepath)), as_root=self.is_rooted)
return boolean(output.strip()) return boolean(output.strip())
def directory_exists(self, filepath): 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, # output from ssh my contain part of the expression in the buffer,
# split out everything except the last word. # split out everything except the last word.
return boolean(output.split()[-1]) # pylint: disable=maybe-no-member return boolean(output.split()[-1]) # pylint: disable=maybe-no-member
@ -642,7 +643,7 @@ class Target(object):
def insmod(self, path): def insmod(self, path):
target_path = self.get_workpath(os.path.basename(path)) target_path = self.get_workpath(os.path.basename(path))
self.push(path, target_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): def extract(self, path, dest=None):
@ -683,7 +684,7 @@ class Target(object):
self.execute('sleep {}'.format(duration), timeout=timeout) self.execute('sleep {}'.format(duration), timeout=timeout)
def read_tree_values_flat(self, path, depth=1, check_exit_code=True): 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, output = self._execute_util(command, as_root=self.is_rooted,
check_exit_code=check_exit_code) check_exit_code=check_exit_code)
@ -731,17 +732,17 @@ class Target(object):
extracted = dest extracted = dest
else: else:
extracted = self.path.dirname(path) 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) self.execute(cmdtext)
return extracted return extracted
def _extract_file(self, path, cmd, dest=None): def _extract_file(self, path, cmd, dest=None):
cmd = '{} ' + cmd # busybox cmd = '{} ' + cmd # busybox
cmdtext = cmd.format(self.busybox, path) cmdtext = cmd.format(quote(self.busybox), quote(path))
self.execute(cmdtext) self.execute(cmdtext)
extracted = self.path.splitext(path)[0] extracted = self.path.splitext(path)[0]
if dest: if dest:
self.execute('mv -f {} {}'.format(extracted, dest)) self.execute('mv -f {} {}'.format(quote(extracted), quote(dest)))
if dest.endswith('/'): if dest.endswith('/'):
extracted = self.path.join(dest, self.path.basename(extracted)) extracted = self.path.join(dest, self.path.basename(extracted))
else: else:
@ -785,7 +786,7 @@ class Target(object):
# It would be nice to use busybox for this, but that means we'd need # 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) # 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, 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 # 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 # 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): def get_pids_of(self, process_name):
"""Returns a list of PIDs of all processes with the specified 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 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() check_exit_code=False).strip().split()
if len(result) >= 2: # at least one row besides the header if len(result) >= 2: # at least one row besides the header
return list(map(int, result[1:])) return list(map(int, result[1:]))
@ -931,7 +932,7 @@ class LinuxTarget(Target):
destpath = self.path.join(self.executables_directory, destpath = self.path.join(self.executables_directory,
with_name and with_name or self.path.basename(filepath)) with_name and with_name or self.path.basename(filepath))
self.push(filepath, destpath) 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 self._installed_binaries[self.path.basename(destpath)] = destpath
return destpath return destpath
@ -947,7 +948,7 @@ class LinuxTarget(Target):
tmpfile = self.tempfile() tmpfile = self.tempfile()
cmd = 'DISPLAY=:0.0 scrot {} && {} date -u -Iseconds' 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) filepath = filepath.format(ts=ts)
self.pull(tmpfile, filepath) self.pull(tmpfile, filepath)
self.remove(tmpfile) self.remove(tmpfile)
@ -1125,7 +1126,7 @@ class AndroidTarget(Target):
if as_root is None: if as_root is None:
as_root = self.needs_su as_root = self.needs_su
try: 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) self.execute(command, timeout=1, as_root=as_root)
except TimeoutError: except TimeoutError:
pass pass
@ -1139,14 +1140,14 @@ class AndroidTarget(Target):
# so we try the new version, and if it fails we use the old version. # so we try the new version, and if it fails we use the old version.
self.ls_command = 'ls -1' self.ls_command = 'ls -1'
try: 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: except TargetStableError:
self.ls_command = 'ls' self.ls_command = 'ls'
def list_directory(self, path, as_root=False): def list_directory(self, path, as_root=False):
if self.ls_command == '': if self.ls_command == '':
self.__setup_list_directory() 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()] return [x.strip() for x in contents.split('\n') if x.strip()]
def install(self, filepath, timeout=None, with_name=None): # pylint: disable=W0221 def install(self, filepath, timeout=None, with_name=None): # pylint: disable=W0221
@ -1194,7 +1195,7 @@ class AndroidTarget(Target):
def capture_screen(self, filepath): def capture_screen(self, filepath):
on_device_file = self.path.join(self.working_directory, 'screen_capture.png') on_device_file = self.path.join(self.working_directory, 'screen_capture.png')
cmd = 'screencap -p {} && {} date -u -Iseconds' 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) filepath = filepath.format(ts=ts)
self.pull(on_device_file, filepath) self.pull(on_device_file, filepath)
self.remove(on_device_file) self.remove(on_device_file)
@ -1285,14 +1286,14 @@ class AndroidTarget(Target):
return output.split() return output.split()
def get_package_version(self, package): 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'): for line in convert_new_lines(output).split('\n'):
if 'versionName' in line: if 'versionName' in line:
return line.split('=', 1)[1] return line.split('=', 1)[1]
return None return None
def get_package_info(self, package): 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'): for entry in output.strip().split('\n'):
rest, entry_package = entry.rsplit('=', 1) rest, entry_package = entry.rsplit('=', 1)
if entry_package != package: if entry_package != package:
@ -1317,13 +1318,13 @@ class AndroidTarget(Target):
if self.get_sdk_version() >= 23: if self.get_sdk_version() >= 23:
flags.append('-g') # Grant all runtime permissions flags.append('-g') # Grant all runtime permissions
self.logger.debug("Replace APK = {}, ADB flags = '{}'".format(replace, ' '.join(flags))) 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: else:
raise TargetStableError('Can\'t install {}: unsupported format.'.format(filepath)) raise TargetStableError('Can\'t install {}: unsupported format.'.format(filepath))
def grant_package_permission(self, package, permission): def grant_package_permission(self, package, permission):
try: try:
return self.execute('pm grant {} {}'.format(package, permission)) return self.execute('pm grant {} {}'.format(quote(package), quote(permission)))
except TargetStableError as e: except TargetStableError as e:
if 'is not a changeable permission type' in e.message: if 'is not a changeable permission type' in e.message:
pass # Ignore if unchangeable pass # Ignore if unchangeable
@ -1353,16 +1354,16 @@ class AndroidTarget(Target):
""" """
Force a re-index of the mediaserver cache for the specified file. 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://' command = 'am broadcast -a android.intent.action.MEDIA_SCANNER_SCAN_FILE -d {}'
self.execute(command + filepath) self.execute(command.format(quote('file://' + filepath)))
def broadcast_media_mounted(self, dirpath, as_root=False): def broadcast_media_mounted(self, dirpath, as_root=False):
""" """
Force a re-index of the mediaserver cache for the specified directory. 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' '-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): def install_executable(self, filepath, with_name=None):
self._ensure_executables_directory_is_writable() self._ensure_executables_directory_is_writable()
@ -1371,14 +1372,14 @@ class AndroidTarget(Target):
on_device_executable = self.path.join(self.executables_directory, executable_name) on_device_executable = self.path.join(self.executables_directory, executable_name)
self.push(filepath, on_device_file) self.push(filepath, on_device_file)
if on_device_file != on_device_executable: 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.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 self._installed_binaries[executable_name] = on_device_executable
return on_device_executable return on_device_executable
def uninstall_package(self, package): 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): def uninstall_executable(self, executable_name):
on_device_executable = self.path.join(self.executables_directory, 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 def dump_logcat(self, filepath, filter=None, append=False, timeout=30): # pylint: disable=redefined-builtin
op = '>>' if append else '>' op = '>>' if append else '>'
filtstr = ' -s {}'.format(filter) if filter else '' filtstr = ' -s {}'.format(quote(filter)) if filter else ''
command = 'logcat -d{} {} {}'.format(filtstr, op, filepath) command = 'logcat -d{} {} {}'.format(filtstr, op, quote(filepath))
adb_command(self.adb_name, command, timeout=timeout) adb_command(self.adb_name, command, timeout=timeout)
def clear_logcat(self): def clear_logcat(self):
@ -1547,8 +1548,8 @@ class AndroidTarget(Target):
if matched: if matched:
entry = sorted(matched, key=lambda x: len(x.mount_point))[-1] entry = sorted(matched, key=lambda x: len(x.mount_point))[-1]
if 'rw' not in entry.options: if 'rw' not in entry.options:
self.execute('mount -o rw,remount {} {}'.format(entry.device, self.execute('mount -o rw,remount {} {}'.format(quote(entry.device),
entry.mount_point), quote(entry.mount_point)),
as_root=True) as_root=True)
else: else:
message = 'Could not find mount point for executables directory {}' message = 'Could not find mount point for executables directory {}'

18
devlib/utils/android.py
View File

@ -235,7 +235,7 @@ class AdbConnection(object):
def push(self, source, dest, timeout=None): def push(self, source, dest, timeout=None):
if timeout is None: if timeout is None:
timeout = self.timeout timeout = self.timeout
command = "push '{}' '{}'".format(source, dest) command = "push {} {}".format(quote(source), quote(dest))
if not os.path.exists(source): if not os.path.exists(source):
raise HostError('No such file "{}"'.format(source)) raise HostError('No such file "{}"'.format(source))
return adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server) 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) command = 'shell {} {}'.format(self.ls_command, source)
output = adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server) output = adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server)
for line in output.splitlines(): 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) adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server)
return 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) return adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server)
# pylint: disable=unused-argument # pylint: disable=unused-argument
@ -285,7 +285,7 @@ class AdbConnection(object):
def fastboot_command(command, timeout=None, device=None): def fastboot_command(command, timeout=None, device=None):
_check_env() _check_env()
target = '-s {}'.format(device) if device else '' target = '-s {}'.format(quote(device)) if device else ''
full_command = 'fastboot {} {}'.format(target, command) full_command = 'fastboot {} {}'.format(target, command)
logger.debug(full_command) logger.debug(full_command)
output, _ = check_output(full_command, timeout, shell=True) 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): 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) fastboot_command(command)
@ -341,7 +341,7 @@ def adb_connect(device, timeout=None, attempts=MAX_ATTEMPTS):
tries += 1 tries += 1
if device: if device:
if "." in device: # Connect is required only for ADB-over-IP 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) logger.debug(command)
output, _ = check_output(command, shell=True, timeout=timeout) output, _ = check_output(command, shell=True, timeout=timeout)
if _ping(device): if _ping(device):
@ -368,7 +368,7 @@ def adb_disconnect(device):
def _ping(device): def _ping(device):
_check_env() _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) command = "adb{} shell \"ls /data/local/tmp > /dev/null\"".format(device_string)
logger.debug(command) logger.debug(command)
result = subprocess.call(command, stderr=subprocess.PIPE, shell=True) 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 # Logcat on older version of android do not support the -e argument
# so fall back to using grep. # so fall back to using grep.
if self.target.get_sdk_version() > 23: if self.target.get_sdk_version() > 23:
logcat_cmd = '{} -e "{}"'.format(logcat_cmd, regexp) logcat_cmd = '{} -e {}'.format(logcat_cmd, quote(regexp))
else: 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) logcat_cmd = get_adb_command(self.target.conn.device, logcat_cmd)

11
devlib/utils/misc.py
View File

@ -35,6 +35,7 @@ import subprocess
import sys import sys
import threading import threading
import wrapt import wrapt
import warnings
from past.builtins import basestring from past.builtins import basestring
@ -416,6 +417,16 @@ def convert_new_lines(text):
""" Convert new lines to a common format. """ """ Convert new lines to a common format. """
return text.replace('\r\n', '\n').replace('\r', '\n') 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): def escape_quotes(text):
"""Escape quotes, and escaped quotes, in the specified text.""" """Escape quotes, and escaped quotes, in the specified text."""

5
devlib/utils/rendering.py
View File

@ -21,6 +21,7 @@ import tempfile
import threading import threading
import time import time
from collections import namedtuple from collections import namedtuple
from pipes import quote
# pylint: disable=redefined-builtin # pylint: disable=redefined-builtin
from devlib.exception import WorkerThreadError, TargetNotRespondingError, TimeoutError from devlib.exception import WorkerThreadError, TargetNotRespondingError, TimeoutError
@ -138,8 +139,8 @@ class SurfaceFlingerFrameCollector(FrameCollector):
self.target.execute('dumpsys SurfaceFlinger --latency-clear ') self.target.execute('dumpsys SurfaceFlinger --latency-clear ')
def get_latencies(self, activity): def get_latencies(self, activity):
cmd = 'dumpsys SurfaceFlinger --latency "{}"' cmd = 'dumpsys SurfaceFlinger --latency {}'
return self.target.execute(cmd.format(activity)) return self.target.execute(cmd.format(quote(activity)))
def list(self): def list(self):
text = self.target.execute('dumpsys SurfaceFlinger --list') text = self.target.execute('dumpsys SurfaceFlinger --list')

50
devlib/utils/ssh.py
View File

@ -26,6 +26,7 @@ import socket
import sys import sys
import time import time
import atexit import atexit
from pipes import quote
# pylint: disable=import-error,wrong-import-position,ungrouped-imports,wrong-import-order # pylint: disable=import-error,wrong-import-position,ungrouped-imports,wrong-import-order
import pexpect import pexpect
@ -39,7 +40,7 @@ from pexpect import EOF, TIMEOUT, spawn
# pylint: disable=redefined-builtin,wrong-import-position # pylint: disable=redefined-builtin,wrong-import-position
from devlib.exception import (HostError, TargetStableError, TargetNotRespondingError, from devlib.exception import (HostError, TargetStableError, TargetNotRespondingError,
TimeoutError, TargetTransientError) 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, from devlib.utils.misc import (escape_single_quotes, escape_double_quotes,
escape_spaces) escape_spaces)
from devlib.utils.types import boolean from devlib.utils.types import boolean
@ -169,7 +170,7 @@ class SshConnection(object):
password_prompt=None, password_prompt=None,
original_prompt=None, original_prompt=None,
platform=None, platform=None,
sudo_cmd="sudo -- sh -c '{}'" sudo_cmd="sudo -- sh -c {}"
): ):
self.host = host self.host = host
self.username = username self.username = username
@ -178,22 +179,18 @@ class SshConnection(object):
self.port = port self.port = port
self.lock = threading.Lock() self.lock = threading.Lock()
self.password_prompt = password_prompt if password_prompt is not None else self.default_password_prompt 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)) logger.debug('Logging in {}@{}'.format(username, host))
timeout = timeout if timeout is not None else self.default_timeout 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) self.conn = ssh_get_shell(host, username, password, self.keyfile, port, timeout, False, None)
atexit.register(self.close) atexit.register(self.close)
def push(self, source, dest, timeout=30): def push(self, source, dest, timeout=30):
dest = '"{}"@"{}":"{}"'.format(escape_double_quotes(self.username), dest = '{}@{}:{}'.format(self.username, self.host, dest)
escape_spaces(escape_double_quotes(self.host)),
escape_spaces(escape_double_quotes(dest)))
return self._scp(source, dest, timeout) return self._scp(source, dest, timeout)
def pull(self, source, dest, timeout=30): def pull(self, source, dest, timeout=30):
source = '"{}"@"{}":"{}"'.format(escape_double_quotes(self.username), source = '{}@{}:{}'.format(self.username, self.host, source)
escape_spaces(escape_double_quotes(self.host)),
escape_spaces(escape_double_quotes(source)))
return self._scp(source, dest, timeout) return self._scp(source, dest, timeout)
def execute(self, command, timeout=None, check_exit_code=True, 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) command = '{} {} {} {}@{} {}'.format(ssh, keyfile_string, port_string, self.username, self.host, command)
logger.debug(command) logger.debug(command)
if self.password: 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) return subprocess.Popen(command, stdout=stdout, stderr=stderr, shell=True)
except EOF: except EOF:
raise TargetNotRespondingError('Connection lost.') raise TargetNotRespondingError('Connection lost.')
@ -310,14 +307,13 @@ class SshConnection(object):
# fails to connect to a device if port is explicitly specified using -P # 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, # 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. # 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 '' port_string = '-P {}'.format(quote(str(self.port))) if (self.port and self.port != 22) else ''
keyfile_string = '-i {}'.format(self.keyfile) if self.keyfile else '' keyfile_string = '-i {}'.format(quote(self.keyfile)) if self.keyfile else ''
command = '{} -r {} {} {} {}'.format(scp, keyfile_string, port_string, source, dest) command = '{} -r {} {} {} {}'.format(scp, keyfile_string, port_string, quote(source), quote(dest))
command_redacted = command command_redacted = command
logger.debug(command) logger.debug(command)
if self.password: if self.password:
command = _give_password(self.password, command) command, command_redacted = _give_password(self.password, command)
command_redacted = command.replace(self.password, '<redacted>')
try: try:
check_output(command, timeout=timeout, shell=True) check_output(command, timeout=timeout, shell=True)
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
@ -456,13 +452,12 @@ class Gem5Connection(TelnetConnection):
if os.path.basename(dest) != filename: if os.path.basename(dest) != filename:
dest = os.path.join(dest, filename) dest = os.path.join(dest, filename)
# Back to the gem5 world # Back to the gem5 world
self._gem5_shell("ls -al {}{}".format(self.gem5_input_dir, filename)) filename = quote(self.gem5_input_dir + filename)
self._gem5_shell("cat '{}''{}' > '{}'".format(self.gem5_input_dir, self._gem5_shell("ls -al {}".format(filename))
filename, self._gem5_shell("cat {} > {}".format(filename, quote(dest)))
dest))
self._gem5_shell("sync") self._gem5_shell("sync")
self._gem5_shell("ls -al {}".format(dest)) self._gem5_shell("ls -al {}".format(quote(dest)))
self._gem5_shell("ls -al {}".format(self.gem5_input_dir)) self._gem5_shell("ls -al {}".format(quote(self.gem5_input_dir)))
logger.debug("Push complete.") logger.debug("Push complete.")
def pull(self, source, dest, timeout=0): #pylint: disable=unused-argument 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.isabs(source):
if os.path.dirname(source) != self.execute('pwd', if os.path.dirname(source) != self.execute('pwd',
check_exit_code=False): check_exit_code=False):
self._gem5_shell("cat '{}' > '{}'".format(filename, self._gem5_shell("cat {} > {}".format(quote(filename),
dest_file)) quote(dest_file)))
self._gem5_shell("sync") self._gem5_shell("sync")
self._gem5_shell("ls -la {}".format(dest_file)) self._gem5_shell("ls -la {}".format(dest_file))
logger.debug('Finished the copy in the simulator') 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 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) output = self.execute(command.format(filepath), as_root=self.is_rooted)
return boolean(output.strip()) return boolean(output.strip())
@ -931,8 +927,10 @@ class AndroidGem5Connection(Gem5Connection):
def _give_password(password, command): def _give_password(password, command):
if not sshpass: if not sshpass:
raise HostError('Must have sshpass installed on the host in order to use password-based auth.') raise HostError('Must have sshpass installed on the host in order to use password-based auth.')
pass_string = "sshpass -p '{}' ".format(password) pass_template = "sshpass -p {} "
return pass_string + command pass_string = pass_template.format(quote(password))
redacted_string = pass_template.format(quote('<redacted>'))
return (pass_string + command, redacted_string + command)
def _check_env(): def _check_env():