escaping: Use pipes.quote instead of escape_*

pipes.quote (also known as shlex.quote in Python3) provides a robust
implementation of quoting and escaping strings before passing them to
POSIX-style shells. Use it instead of the escape_* functions to simplify
quoting inside devlib.

devlib/target.py

@@ -26,6 +26,7 @@ import threading
 import xml.dom.minidom
 import copy
 from collections import namedtuple, defaultdict
+from pipes import quote
 
 from devlib.host import LocalConnection, PACKAGE_BIN_DIRECTORY
 from devlib.module import get_module
@@ -36,7 +37,7 @@ from devlib.exception import (DevlibTransientError, TargetStableError,
 from devlib.utils.ssh import SshConnection
 from devlib.utils.android import AdbConnection, AndroidProperties, LogcatMonitor, adb_command, adb_disconnect, INTENT_FLAGS
 from devlib.utils.misc import memoized, isiterable, convert_new_lines
-from devlib.utils.misc import commonprefix, escape_double_quotes, merge_lists
+from devlib.utils.misc import commonprefix, merge_lists
 from devlib.utils.misc import ABI_MAP, get_cpu_name, ranges_to_list
 from devlib.utils.types import integer, boolean, bitmask, identifier, caseless_string, bytes_regex
 
@@ -555,7 +556,7 @@ class Target(object):
         raise IOError('No usable temporary filename found')
 
     def remove(self, path, as_root=False):
-        self.execute('rm -rf "{}"'.format(escape_double_quotes(path)), as_root=as_root)
+        self.execute('rm -rf {}'.format(quote(path)), as_root=as_root)
 
     # misc
     def core_cpus(self, core):
@@ -891,7 +892,7 @@ class LinuxTarget(Target):
         pass
 
     def kick_off(self, command, as_root=False):
-        command = 'sh -c "{}" 1>/dev/null 2>/dev/null &'.format(escape_double_quotes(command))
+        command = 'sh -c {} 1>/dev/null 2>/dev/null &'.format(quote(command))
         return self.conn.execute(command, as_root=as_root)
 
     def get_pids_of(self, process_name):
@@ -925,7 +926,7 @@ class LinuxTarget(Target):
             return filtered_result
 
     def list_directory(self, path, as_root=False):
-        contents = self.execute('ls -1 "{}"'.format(escape_double_quotes(path)), as_root=as_root)
+        contents = self.execute('ls -1 {}'.format(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
@@ -1522,13 +1523,13 @@ class AndroidTarget(Target):
             if it is already running
         :type force_new: bool
         """
-        cmd = 'am start -a android.intent.action.VIEW -d "{}"'
+        cmd = 'am start -a android.intent.action.VIEW -d {}'
 
         if force_new:
             cmd = cmd + ' -f {}'.format(INTENT_FLAGS['ACTIVITY_NEW_TASK'] |
                                         INTENT_FLAGS['ACTIVITY_CLEAR_TASK'])
 
-        self.execute(cmd.format(escape_double_quotes(url)))
+        self.execute(cmd.format(quote(url)))
 
     def homescreen(self):
         self.execute('am start -a android.intent.action.MAIN -c android.intent.category.HOME')

devlib/utils/android.py

@@ -28,10 +28,10 @@ import tempfile
 import subprocess
 from collections import defaultdict
 import pexpect
+from pipes import quote
 
 from devlib.exception import TargetTransientError, TargetStableError, HostError
 from devlib.utils.misc import check_output, which, ABI_MAP
-from devlib.utils.misc import escape_single_quotes, escape_double_quotes
 
 
 logger = logging.getLogger('android')
@@ -383,7 +383,7 @@ def adb_shell(device, command, timeout=None, check_exit_code=False,
               as_root=False, adb_server=None):  # NOQA
     _check_env()
     if as_root:
-        command = 'echo \'{}\' | su'.format(escape_single_quotes(command))
+        command = 'echo {} | su'.format(quote(command))
     device_part = []
     if adb_server:
         device_part = ['-H', adb_server]
@@ -443,9 +443,9 @@ def adb_background_shell(device, command,
     """Runs the sepcified command in a subprocess, returning the the Popen object."""
     _check_env()
     if as_root:
-        command = 'echo \'{}\' | su'.format(escape_single_quotes(command))
+        command = 'echo {} | su'.format(quote(command))
     device_string = ' -s {}'.format(device) if device else ''
-    full_command = 'adb{} shell "{}"'.format(device_string, escape_double_quotes(command))
+    full_command = 'adb{} shell {}'.format(device_string, quote(command))
     logger.debug(full_command)
     return subprocess.Popen(full_command, stdout=stdout, stderr=stderr, shell=True)
 

devlib/utils/misc.py

@@ -429,22 +429,38 @@ def sanitize_cmd_template(cmd):
     return cmd
 
 def escape_quotes(text):
-    """Escape quotes, and escaped quotes, in the specified text."""
+    """
+    Escape quotes, and escaped quotes, in the specified text.
+
+    .. note:: :func:`pipes.quote` should be favored where possible.
+    """
     return re.sub(r'\\("|\')', r'\\\\\1', text).replace('\'', '\\\'').replace('\"', '\\\"')
 
 
 def escape_single_quotes(text):
-    """Escape single quotes, and escaped single quotes, in the specified text."""
+    """
+    Escape single quotes, and escaped single quotes, in the specified text.
+
+    .. note:: :func:`pipes.quote` should be favored where possible.
+    """
     return re.sub(r'\\("|\')', r'\\\\\1', text).replace('\'', '\'\\\'\'')
 
 
 def escape_double_quotes(text):
-    """Escape double quotes, and escaped double quotes, in the specified text."""
+    """
+    Escape double quotes, and escaped double quotes, in the specified text.
+
+    .. note:: :func:`pipes.quote` should be favored where possible.
+    """
     return re.sub(r'\\("|\')', r'\\\\\1', text).replace('\"', '\\\"')
 
 
 def escape_spaces(text):
-    """Escape spaces in the specified text"""
+    """
+    Escape spaces in the specified text
+
+    .. note:: :func:`pipes.quote` should be favored where possible.
+    """
     return text.replace(' ', '\ ')
 
 

devlib/utils/ssh.py

@@ -41,8 +41,6 @@ from pexpect import EOF, TIMEOUT, spawn
 from devlib.exception import (HostError, TargetStableError, TargetNotRespondingError,
                               TimeoutError, TargetTransientError)
 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
 
 
@@ -265,7 +263,7 @@ class SshConnection(object):
             # As we're already root, there is no need to use sudo.
             as_root = False
         if as_root:
-            command = self.sudo_cmd.format(escape_single_quotes(command))
+            command = self.sudo_cmd.format(quote(command))
             if log:
                 logger.debug(command)
             self.conn.sendline(command)
@@ -758,7 +756,7 @@ class Gem5Connection(TelnetConnection):
         gem5_logger.debug("gem5_shell command: {}".format(command))
 
         if as_root:
-            command = 'echo "{}" | su'.format(escape_double_quotes(command))
+            command = 'echo {} | su'.format(quote(command))
 
         # Send the actual command
         self.conn.send("{}\n".format(command))