Repository: Azure/WALinuxAgent
Pull request #2086 (2020-11-23 19:18:10)

Add support for redirection to shellutil.run_command
Extended shellutil.run_command to support input/output redirection.

Added shellutil.run_pipe to execute command pipes.


Patch:
diff --git a/azurelinuxagent/common/logcollector.py b/azurelinuxagent/common/logcollector.py
index 73f659b2..930277bf 100644
--- a/azurelinuxagent/common/logcollector.py
+++ b/azurelinuxagent/common/logcollector.py
@@ -115,7 +115,7 @@ class LogCollector(object):  # pylint: disable=R0903
             return ustr(output, encoding="utf-8", errors="backslashreplace")
 
         try:
-            process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=stdout, stderr=subprocess.PIPE, shell=False)
+            process = subprocess.Popen(command, stdout=stdout, stderr=subprocess.PIPE, shell=False)
             stdout, stderr = process.communicate()
             return_code = process.returncode
         except Exception as e:  # pylint: disable=C0103
diff --git a/azurelinuxagent/common/osutil/default.py b/azurelinuxagent/common/osutil/default.py
index 681961b5..d7178b68 100644
--- a/azurelinuxagent/common/osutil/default.py
+++ b/azurelinuxagent/common/osutil/default.py
@@ -1472,7 +1472,7 @@ class DefaultOSUtil(object):  # pylint: disable=R0904
     def _run_command_raising_OSUtilError(cmd, err_msg, cmd_input=None):  # pylint: disable=C0103
         # This method runs shell command using the new secure shellutil.run_command and raises OSUtilErrors on failures.
         try:
-            return shellutil.run_command(cmd, log_error=True, cmd_input=cmd_input)
+            return shellutil.run_command(cmd, log_error=True, input=cmd_input)
         except shellutil.CommandError as e:  # pylint: disable=C0103
             raise OSUtilError(
                 "{0}, Retcode: {1}, Output: {2}, Error: {3}".format(err_msg, e.returncode, e.stdout, e.stderr))
diff --git a/azurelinuxagent/common/utils/cryptutil.py b/azurelinuxagent/common/utils/cryptutil.py
index 028372e3..44ecf6ab 100644
--- a/azurelinuxagent/common/utils/cryptutil.py
+++ b/azurelinuxagent/common/utils/cryptutil.py
@@ -81,46 +81,22 @@ class CryptUtil(object):
         elif not os.path.exists(trans_prv_file):
             raise IOError(errno.ENOENT, "File not found", trans_prv_file)
         else:
-            first_cmd = [self.openssl_cmd, "cms", "-decrypt", "-in", p7m_file, "-inkey",
-                trans_prv_file, "-recip", trans_cert_file]
-            second_cmd = [self.openssl_cmd, "pkcs12", "-nodes", "-password", "pass:", 
-                "-out", pem_file]
-
-            first_proc = subprocess.Popen(first_cmd, stdout=subprocess.PIPE)
-
-            second_proc = subprocess.Popen(second_cmd, stdin=first_proc.stdout,
-                stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-            first_proc.stdout.close()  # see https://docs.python.org/2/library/subprocess.html#replacing-shell-pipeline
-            stdout, stderr = second_proc.communicate()
-
-            if second_proc.returncode != 0:
-                stdout = ustr(stdout, encoding='utf-8', errors="backslashreplace") if stdout else ""
-                stderr =  ustr(stderr, encoding='utf-8', errors="backslashreplace") if stderr else ""
-
-                msg = "Failed to decrypt {0}\n[stdout]\n{1}\n\n[stderr]\n{2}\n"\
-                    .format(p7m_file, stdout, stderr)
-                logger.error(msg)
-            
+            try:
+                shellutil.run_pipe([
+                    [self.openssl_cmd, "cms", "-decrypt", "-in", p7m_file, "-inkey", trans_prv_file, "-recip", trans_cert_file],
+                    [self.openssl_cmd, "pkcs12", "-nodes", "-password", "pass:", "-out", pem_file]])
+            except shellutil.CommandError as command_error:
+                logger.error("Failed to decrypt {0} (return code: {1})\n[stdout]\n{2}\n[stderr]\n{3}",
+                    p7m_file, command_error.returncode, command_error.stdout, command_error.stderr)
 
     def crt_to_ssh(self, input_file, output_file):
         with open(output_file, "ab") as file_out:
             cmd = ["ssh-keygen", "-i", "-m", "PKCS8", "-f", input_file]
 
             try:
-                keygen_proc = subprocess.Popen(cmd, stdout=file_out)
-                keygen_proc.wait()
-
-                if keygen_proc.returncode != 0:
-                    msg = u"Command: [{0}], return code: [{1}]" \
-                        .format(cmd, keygen_proc.returncode)
-                    logger.error(msg)
-                
-            except Exception as exception:
-                msg = u"Exception on Command: [{0}]. exception={1}" \
-                    .format(cmd, exception)
-
-
-
+                shellutil.run_command(cmd, stdout=file_out, log_error=True)
+            except shellutil.CommandError:
+                pass  # nothing to do; the error is already logged
 
     def asn1_to_ssh(self, pubkey):
         lines = pubkey.split("\n")
@@ -179,12 +155,9 @@ class CryptUtil(object):
         try:
             decoded = base64.b64decode(encrypted_password)
             args = DECRYPT_SECRET_CMD.format(self.openssl_cmd, private_key).split(' ')
-            p = subprocess.Popen(args, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.STDOUT)  # pylint: disable=C0103
-            p.stdin.write(decoded)
-            output = p.communicate()[0]
-            retcode = p.poll()
-            if retcode:
-                raise subprocess.CalledProcessError(retcode, "openssl cms -decrypt", output=output)
+            output = shellutil.run_command(args, input=decoded, stderr=subprocess.STDOUT, encode_input=False, encode_output=False)
             return output.decode('utf-16')
+        except shellutil.CommandError as command_error:
+            raise subprocess.CalledProcessError(command_error.returncode, "openssl cms -decrypt", output=command_error.stdout)
         except Exception as e:  # pylint: disable=C0103
             raise CryptError("Error decoding secret", e)
diff --git a/azurelinuxagent/common/utils/shellutil.py b/azurelinuxagent/common/utils/shellutil.py
index 3be6b5ae..24ab0008 100644
--- a/azurelinuxagent/common/utils/shellutil.py
+++ b/azurelinuxagent/common/utils/shellutil.py
@@ -18,6 +18,7 @@
 #
 
 import subprocess
+import tempfile
 
 import azurelinuxagent.common.logger as logger
 from azurelinuxagent.common.future import ustr
@@ -97,12 +98,10 @@ def run_get_output(cmd, chk_err=True, log_cmd=True, expected_errors=None):
     if log_cmd:
         logger.verbose(u"Command: [{0}]", cmd)
     try:
-        output = subprocess.check_output(cmd,
-                                         stderr=subprocess.STDOUT,
-                                         shell=True)
-        output = _encode_command_output(output)
+        output = subprocess.check_output(cmd, stderr=subprocess.STDOUT, shell=True)
+        output = __encode_command_output(output)
     except subprocess.CalledProcessError as e:  # pylint: disable=C0103
-        output = _encode_command_output(e.output)
+        output = __encode_command_output(e.output)
 
         if chk_err:
             msg = u"Command: [{0}], " \
@@ -121,13 +120,35 @@ def run_get_output(cmd, chk_err=True, log_cmd=True, expected_errors=None):
     return 0, output
 
 
-def _encode_command_output(output):
-    return ustr(output, encoding='utf-8', errors="backslashreplace")
+def __format_command(command):
+    """
+    Formats the command taken by run_command/run_pipe.
+
+    Examples:
+        > __format_command("sort")
+        'sort'
+        > __format_command(["sort", "-u"])
+        'sort -u'
+        > __format_command([["sort"], ["unique", "-n"]])
+        'sort | unique -n'
+    """
+    if isinstance(command, list):
+        if command and isinstance(command[0], list):
+            return " | ".join([" ".join(cmd) for cmd in command])
+        return " ".join(command)
+    return command
+
+
+def __encode_command_output(output):
+    """
+    Encodes the stdout/stderr returned by subprocess.communicate()
+    """
+    return ustr(output if output is not None else b'', encoding='utf-8', errors="backslashreplace")
 
 
 class CommandError(Exception):
     """
-    Exception raised by run_command when the command returns an error
+    Exception raised by run_command/run_pipe when the command returns an error
     """
     @staticmethod
     def _get_message(command, return_code, stderr):
@@ -142,47 +163,150 @@ class CommandError(Exception):
         self.stderr = stderr
 
 
-def run_command(command, log_error=False, cmd_input=None):
+def __run_command(command_action, command, log_error, encode_output):
     """
-        Executes the given command and returns its stdout as a string. If cmd_input is specified, then we pass the cmd_input
-        to stdin and execute the command. Currently we only support string input for stdin.
-        If there are any errors executing the command it logs details about the failure and raises a RunCommandException;
-        if 'log_error' is True, it also logs details about the error.
+    Executes the given command_action and returns its stdout. The command_action is a function that executes a command/pipe
+    and returns its exit code, stdout, and stderr.
 
-        Note: This is the preferred method to execute shell commands over `azurelinuxagent.common.utils.shellutil.run` function.
-    """
-    def format_command(cmd):
-        return " ".join(cmd) if isinstance(cmd, list) else command
+    If there are any errors executing the command it raises a RunCommandException; if 'log_error'
+    is True, it also logs details about the error.
 
-    # Currently we only support PIPE for stdin/stdout/stderr, but acceptable options as per python docs are -
-    # PIPE, an existing file descriptor (a positive integer), an existing file object, and None
-    stdin = subprocess.PIPE if cmd_input else None
+    If encode_output is True the stdout is returned as a string, otherwise it is returned as a bytes object.
+    """
     try:
-        # Starting Python 3.4+, you need to encode the string, i.e. you need to pass Bytes to the input rather than
-        # string to process.communicate()
-        process_input = cmd_input.encode() if cmd_input else None
+        return_code, stdout, stderr = command_action()
 
-        process = subprocess.Popen(command, stdin=stdin, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False)
-        stdout, stderr = process.communicate(input=process_input)
-        returncode = process.returncode
-    except Exception as e:  # pylint: disable=C0103
+        if encode_output:
+            stdout = __encode_command_output(stdout)
+            stderr = __encode_command_output(stderr)
+
+        if return_code != 0:
+            if log_error:
+                logger.error(
+                    "Command: [{0}], return code: [{1}], stdout: [{2}] stderr: [{3}]",
+                    __format_command(command),
+                    return_code,
+                    stdout,
+                    stderr)
+            raise CommandError(command=__format_command(command), return_code=return_code, stdout=stdout, stderr=stderr)
+
+        return stdout
+
+    except CommandError:
+        raise
+    except Exception as exception:
         if log_error:
-            logger.error(u"Command [{0}] raised unexpected exception: [{1}]", format_command(command), ustr(e))
+            logger.error(u"Command [{0}] raised unexpected exception: [{1}]", __format_command(command), ustr(exception))
         raise
 
-    if returncode != 0:
-        encoded_stdout = _encode_command_output(stdout)
-        encoded_stderr = _encode_command_output(stderr)
-        if log_error:
-            logger.error(
-                "Command: [{0}], return code: [{1}], stdout: [{2}] stderr: [{3}]",
-                format_command(command),
-                returncode,
-                encoded_stdout,
-                encoded_stderr)
-        raise CommandError(command=command, return_code=returncode, stdout=encoded_stdout, stderr=encoded_stderr)
-
-    return _encode_command_output(stdout)
+
+# W0622: Redefining built-in 'input'  -- disabled: the parameter name mimics subprocess.communicate()
+# R0913: Too many arguments (8/5) -- disabled: the parameter list mimics subprocess.Popen()/communicate()
+def run_command(command, input=None, stdin=None, stdout=subprocess.PIPE, stderr=subprocess.PIPE, log_error=False, encode_input=True, encode_output=True):  # pylint:disable=W0622,R0913
+    """
+        Executes the given command and returns its stdout.
+
+        If there are any errors executing the command it raises a RunCommandException; if 'log_error'
+        is True, it also logs details about the error.
+
+        If encode_output is True the stdout is returned as a string, otherwise it is returned as a bytes object.
+
+        This function is a thin wrapper around Popen/communicate in the subprocess module:
+           * The 'input' parameter corresponds to the same parameter in communicate
+           * The 'stdin' parameter corresponds to the same parameters in Popen
+           * Only one of 'input' and 'stdin' can be specified
+           * The 'stdout' and 'stderr' parameters correspond to the same parameters in Popen, except that they
+             default to subprocess.PIPE instead of None
+           * If the output of the command is redirected using the 'stdout' or 'stderr' parameters (i.e. if the
+             value for these parameters is anything other than the default (subprocess.PIPE)), then the corresponding
+             values returned by this function or the CommandError exception will be empty strings.
+
+        Note: This is the preferred method to execute shell commands over `azurelinuxagent.common.utils.shellutil.run` function.
+    """
+    if input is not None and stdin is not None:
+        raise ValueError("The input and stdin arguments are mutually exclusive")
+
+    def command_action():
+        popen_stdin = communicate_input = None
+        if input is not None:
+            popen_stdin = subprocess.PIPE
+            communicate_input = input.encode() if encode_input and isinstance(input, str) else input  # communicate() needs an array of bytes
+        if stdin is not None:
+            popen_stdin = stdin
+            communicate_input = None
+
+        process = subprocess.Popen(command, stdin=popen_stdin, stdout=stdout, stderr=stderr, shell=False)
+
+        command_stdout, command_stderr = process.communicate(input=communicate_input)
+
+        return process.returncode, command_stdout, command_stderr
+
+    return __run_command(command_action=command_action, command=command, log_error=log_error, encode_output=encode_output)
+
+
+# R0913: Too many arguments (7/5) -- disabled: the parameter list mimics subprocess.Popen()
+def run_pipe(pipe, stdin=None, stdout=subprocess.PIPE, stderr=subprocess.PIPE, log_error=False, encode_output=True):   # pylint:disable=R0913
+    """
+        Executes the given commands as a pipe and returns its stdout as a string.
+
+        The pipe is a list of commands, which in turn are a list of strings, e.g.
+
+            [["sort"], ["uniq", "-n"]] represents 'sort | unique -n'
+
+        If there are any errors executing the command it raises a RunCommandException; if 'log_error'
+        is True, it also logs details about the error.
+
+        If encode_output is True the stdout is returned as a string, otherwise it is returned as a bytes object.
+
+        This function is a thin wrapper around Popen/communicate in the subprocess module:
+           * The 'stdin' parameter is used as input for the first command in the pipe
+           * The 'stdout', and 'stderr' can be used to redirect the output of the pipe
+           * If the output of the pipe is redirected using the 'stdout' or 'stderr' parameters (i.e. if the
+             value for these parameters is anything other than the default (subprocess.PIPE)), then the corresponding
+             values returned by this function or the CommandError exception will be empty strings.
+    """
+    if len(pipe) < 2:
+        raise ValueError("The pipe must consist of at least 2 commands")
+
+    def command_action():
+        stderr_file = None
+
+        try:
+            popen_stdin = stdin
+            # If stderr is subprocess.PIPE each call to Popen would create a new pipe. We want to collect the stderr of all the
+            # commands in the pipe so we replace stderr with a temporary file that we read once the pipe completes.
+            if stderr == subprocess.PIPE:
+                stderr_file = tempfile.TemporaryFile()
+                popen_stderr = stderr_file
+            else:
+                popen_stderr = stderr
+
+            processes = []
+            i = 0
+            while i < len(pipe) - 1:
+                processes.append(subprocess.Popen(pipe[i], stdin=popen_stdin, stdout=subprocess.PIPE, stderr=popen_stderr))
+                popen_stdin = processes[i].stdout
+                i += 1
+
+            processes.append(subprocess.Popen(pipe[i], stdin=popen_stdin, stdout=stdout, stderr=popen_stderr))
+
+            i = 0
+            while i < len(processes) - 1:
+                processes[i].stdout.close()  # see https://docs.python.org/2/library/subprocess.html#replacing-shell-pipeline
+                i += 1
+
+            pipe_stdout, pipe_stderr = processes[i].communicate()
+
+            if stderr_file is not None:
+                stderr_file.seek(0)
+                pipe_stderr = stderr_file.read()
+
+            return processes[i].returncode, pipe_stdout, pipe_stderr
+        finally:
+            if stderr_file is not None:
+                stderr_file.close()
+
+    return __run_command(command_action=command_action, command=pipe, log_error=log_error, encode_output=encode_output)
 
 
 def quote(word_list):


Test patch:
diff --git a/tests/ga/test_extension.py b/tests/ga/test_extension.py
index 1af40fc2..9a7af0e4 100644
--- a/tests/ga/test_extension.py
+++ b/tests/ga/test_extension.py
@@ -2168,8 +2168,10 @@ class TestExtension(ExtensionTestCase):
                 printenv | grep ConfigSequenceNumber
                 """
 
-        base_dir = os.path.join(conf.get_lib_dir(), 'OSTCExtensions.ExampleHandlerLinux-1.0.0', test_file_name)
-        self.create_script(test_file_name, test_file, base_dir)
+        base_dir = os.path.join(conf.get_lib_dir(), 'OSTCExtensions.ExampleHandlerLinux-1.0.0')
+        if not os.path.exists(base_dir):
+            os.mkdir(base_dir)
+        self.create_script(os.path.join(base_dir, test_file_name), test_file)
 
         test_data = mockwiredata.WireProtocolData(mockwiredata.DATA_FILE_EXT_SINGLE)
         exthandlers_handler, protocol = self._create_mock(test_data, *args)  # pylint: disable=unused-variable,no-value-for-parameter
@@ -2192,8 +2194,10 @@ class TestExtension(ExtensionTestCase):
             test_data.ext_conf = test_data.ext_conf.replace('seqNo="0"', 'seqNo="1"')
             test_data.manifest = test_data.manifest.replace('1.0.0', '1.0.1')
             expected_seq_no = 1
-            base_dir = os.path.join(conf.get_lib_dir(), 'OSTCExtensions.ExampleHandlerLinux-1.0.1', test_file_name)
-            self.create_script(test_file_name, test_file, base_dir)
+            base_dir = os.path.join(conf.get_lib_dir(), 'OSTCExtensions.ExampleHandlerLinux-1.0.1')
+            if not os.path.exists(base_dir):
+                os.mkdir(base_dir)
+            self.create_script(os.path.join(base_dir, test_file_name), test_file)
 
             with patch.object(ExtHandlerInstance, 'report_event') as mock_report_event:
                 exthandlers_handler.run()
@@ -2230,13 +2234,17 @@ class TestExtension(ExtensionTestCase):
             exit %s
         """ % exit_code
 
-        error_dir = os.path.join(conf.get_lib_dir(), 'OSTCExtensions.ExampleHandlerLinux-1.0.0', test_error_file_name)
-        self.create_script(test_error_file_name, test_error_content, error_dir)
+        error_dir = os.path.join(conf.get_lib_dir(), 'OSTCExtensions.ExampleHandlerLinux-1.0.0')
+        if not os.path.exists(error_dir):
+            os.mkdir(error_dir)
+        self.create_script(os.path.join(error_dir, test_error_file_name), test_error_content)
 
         test_data, exthandlers_handler, protocol = self._set_up_update_test_and_update_gs(Mock(), *args)  # pylint: disable=unused-variable
 
-        base_dir = os.path.join(conf.get_lib_dir(), 'OSTCExtensions.ExampleHandlerLinux-1.0.1', test_file_name)
-        self.create_script(test_file_name, test_file, base_dir)
+        base_dir = os.path.join(conf.get_lib_dir(), 'OSTCExtensions.ExampleHandlerLinux-1.0.1')
+        if not os.path.exists(base_dir):
+            os.mkdir(base_dir)
+        self.create_script(os.path.join(base_dir, test_file_name), test_file)
 
         with patch("azurelinuxagent.ga.exthandlers.ExtHandlerInstance.load_manifest", return_value=manifest):
             with patch.object(ExtHandlerInstance, 'report_event') as mock_report_event:
diff --git a/tests/ga/test_exthandlers.py b/tests/ga/test_exthandlers.py
index 58305539..fdd71c8f 100644
--- a/tests/ga/test_exthandlers.py
+++ b/tests/ga/test_exthandlers.py
@@ -349,7 +349,8 @@ class LaunchCommandTestCase(AgentTestCase):
         stdout = "stdout" * 5
         stderr = "stderr" * 5
 
-        command = self.create_script("produce_output.py", '''
+        command = "produce_output.py"
+        self.create_script(os.path.join(self.ext_handler_instance.get_base_dir(), command), '''
 import sys
 
 sys.stdout.write("{0}")
@@ -380,7 +381,8 @@ sys.stderr.write("{1}")
         signal_file = os.path.join(self.tmp_dir, "signal_file.txt")
 
         # the test command produces some output then goes into an infinite loop
-        command = self.create_script("produce_output_then_hang.py", '''
+        command = "produce_output_then_hang.py"
+        self.create_script(os.path.join(self.ext_handler_instance.get_base_dir(), command), '''
 import sys
 import time
 
@@ -446,7 +448,8 @@ with open("{2}", "w") as file:
         stderr = "stderr" * 3
         exit_code = 99
 
-        command = self.create_script("fail.py", '''
+        command = "fail.py"
+        self.create_script(os.path.join(self.ext_handler_instance.get_base_dir(), command), '''
 import sys
 
 sys.stdout.write("{0}")
@@ -468,7 +471,8 @@ exit({2})
         stdout = "stdout"
         stderr = "stderr"
 
-        command = self.create_script("start_child_process.py", '''
+        command = "start_child_process.py"
+        self.create_script(os.path.join(self.ext_handler_instance.get_base_dir(), command), '''
 import os
 import sys
 import time
@@ -503,7 +507,8 @@ else:
         # the child process uses the signal file to indicate it has produced output
         signal_file = os.path.join(self.tmp_dir, "signal_file.txt")
 
-        command = self.create_script("start_child_with_output.py", '''
+        command = "start_child_with_output.py"
+        self.create_script(os.path.join(self.ext_handler_instance.get_base_dir(), command), '''
 import os
 import sys
 import time
@@ -544,7 +549,8 @@ else:
         child_stdout = "CHILD STDOUT"
         child_stderr = "CHILD STDERR"
 
-        command = self.create_script("start_child_that_fails.py", '''
+        command = "start_child_that_fails.py"
+        self.create_script(os.path.join(self.ext_handler_instance.get_base_dir(), command), '''
 import os
 import sys
 import time
@@ -573,7 +579,8 @@ else:
         # file used to verify the command completed successfully
         signal_file = os.path.join(self.tmp_dir, "signal_file.txt")
 
-        command = self.create_script("create_file.py", '''
+        command = "create_file.py"
+        self.create_script(os.path.join(self.ext_handler_instance.get_base_dir(), command), '''
 open("{0}", "w").close()
 
 '''.format(signal_file))
@@ -590,7 +597,8 @@ open("{0}", "w").close()
         stderr = "STDERR"
 
         # the test script mimics the redirection done by the Custom Script extension
-        command = self.create_script("produce_output", '''
+        command = "produce_output"
+        self.create_script(os.path.join(self.ext_handler_instance.get_base_dir(), command), '''
 exec &> {0}
 echo {1}
 >&2 echo {2}
@@ -609,7 +617,8 @@ echo {1}
         stdout = "STDOUT"
         stderr = "STDERR"
 
-        command = self.create_script("produce_long_output.py", '''
+        command = "produce_long_output.py"
+        self.create_script(os.path.join(self.ext_handler_instance.get_base_dir(), command), '''
 import sys
 
 sys.stdout.write( "{0}" * {1})
@@ -623,7 +632,8 @@ sys.stderr.write( "{2}" * {3})
         self.assertIn(stderr, output)
 
     def test_it_should_read_only_the_head_of_large_outputs(self):
-        command = self.create_script("produce_long_output.py", '''
+        command = "produce_long_output.py"
+        self.create_script(os.path.join(self.ext_handler_instance.get_base_dir(), command), '''
 import sys
 
 sys.stdout.write("O" * 5 * 1024 * 1024)
@@ -651,7 +661,8 @@ sys.stderr.write("E" * 5 * 1024 * 1024)
         self.assertLessEqual(len(stderr), TELEMETRY_MESSAGE_MAX_LEN)
 
     def test_it_should_handle_errors_while_reading_the_command_output(self):
-        command = self.create_script("produce_output.py", '''
+        command = "produce_output.py"
+        self.create_script(os.path.join(self.ext_handler_instance.get_base_dir(), command), '''
 import sys
 
 sys.stdout.write("STDOUT")
@@ -684,7 +695,8 @@ sys.stderr.write("STDERR")
             printenv | grep -E '(%s)'
         """ % '|'.join(helper_env_vars.keys())
 
-        test_file = self.create_script('printHelperEnvironments.sh', command)
+        test_file = 'printHelperEnvironments.sh'
+        self.create_script(os.path.join(self.ext_handler_instance.get_base_dir(), test_file), command)
 
         with patch("subprocess.Popen", wraps=subprocess.Popen) as patch_popen:
             output = ext_handler_instance.launch_command(test_file)
diff --git a/tests/tools.py b/tests/tools.py
index f57b67f6..e6903677 100644
--- a/tests/tools.py
+++ b/tests/tools.py
@@ -421,32 +421,20 @@ class AgentTestCase(unittest.TestCase):
             fileutil.write_file(f, "faux content")
             time.sleep(with_sleep)
 
-    def create_script(self, file_name, contents, file_path=None):
+    @staticmethod
+    def create_script(script_file, contents):
         """
-        Creates an executable script with the given contents.
-        If file_name ends with ".py", it creates a Python3 script, otherwise it creates a bash script
-        :param file_name: The name of the file to create the script with
-        :param contents: Contents of the script file
-        :param file_path: The path of the file where to create it in (we use /tmp/ by default)
-        :return:
+        Creates an executable script with the given contents. If file ends with ".py", it creates a Python3 script,
+        otherwise it creates a bash script.
         """
-        if not file_path:
-            file_path = os.path.join(self.tmp_dir, file_name)
-
-        directory = os.path.dirname(file_path)
-        if not os.path.exists(directory):
-            os.mkdir(directory)
-
-        with open(file_path, "w") as script:
-            if file_name.endswith(".py"):
+        with open(script_file, "w") as script:
+            if script_file.endswith(".py"):
                 script.write("#!/usr/bin/env python3\n")
             else:
                 script.write("#!/usr/bin/env bash\n")
             script.write(contents)
 
-        os.chmod(file_path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
-
-        return file_name
+        os.chmod(script_file, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
 
 
 def load_data(name):
diff --git a/tests/utils/test_shell_util.py b/tests/utils/test_shell_util.py
index ac573254..fdcf5afe 100644
--- a/tests/utils/test_shell_util.py
+++ b/tests/utils/test_shell_util.py
@@ -15,8 +15,11 @@
 #
 # Requires Python 2.6+ and Openssl 1.0+
 #
+import os
+import tempfile
 import unittest
 
+from azurelinuxagent.common.future import ustr
 import azurelinuxagent.common.utils.shellutil as shellutil
 from tests.tools import AgentTestCase, patch
 
@@ -127,41 +130,97 @@ class RunGetOutputTestCase(AgentTestCase):
         self.assertEqual(mock_logger.warn.call_count, 0)
 
 
-class RunCommandTestCase(AgentTestCase):
+# R0904: Too many public methods (24/20)  -- disabled: each method is a unit test
+class RunCommandTestCase(AgentTestCase):  # pylint: disable=R0904
+    """
+    Tests for shellutil.run_command/run_pipe
+    """
+    def __create_tee_script(self, return_code=0):
+        """
+        Creates a Python script that tees its stdin to stdout and stderr
+        """
+        tee_script = os.path.join(self.tmp_dir, "tee.py")
+
+        AgentTestCase.create_script(tee_script, """
+import sys
+
+for line in sys.stdin:
+    sys.stdout.write(line)
+    sys.stderr.write(line)
+exit({0})
+    """.format(return_code))
+
+        return tee_script
+
     def test_run_command_should_execute_the_command(self):
         command = ["echo", "-n", "A TEST STRING"]
         ret = shellutil.run_command(command)
         self.assertEqual(ret, "A TEST STRING")
 
-    def test_run_command_should_raise_an_exception_when_the_command_fails(self):
-        command = ["ls", "-d", "/etc", "nonexistent_file"]
+    def test_run_pipe_should_execute_a_pipe_with_two_commands(self):
+        # Output the same string 3 times and then remove duplicates
+        test_string = "A TEST STRING\n"
+        pipe = [["echo", "-n", "-e", test_string * 3], ["uniq"]]
+
+        output = shellutil.run_pipe(pipe)
+
+        self.assertEqual(output, test_string)
 
+    def test_run_pipe_should_execute_a_pipe_with_more_than_two_commands(self):
+        #
+        # The test pipe splits the output of "ls" in lines and count the number of lines.
+        #
+        # Sample output of "ls":
+        #     drwxrwxr-x 13 nam nam 4096 Nov 13 16:54 .
+        #
+        pipe = [["ls", "-ld", "."], ["sed", "s/ /\\n/g"], ["wc", "-l"]]
+
+        output = shellutil.run_pipe(pipe)
+
+        self.assertEqual(int(output), 9)
+
+    def __it_should_raise_an_exception_when_the_command_fails(self, action):
         with self.assertRaises(shellutil.CommandError) as context_manager:
-            shellutil.run_command(command)
+            action()
 
         exception = context_manager.exception
-        self.assertIn("'ls' failed: 2", str(exception))
-        self.assertIn("No such file or directory", str(exception))
-        self.assertEqual(exception.stdout, "/etc\n")
-        self.assertIn("No such file or directory", exception.stderr)
-        self.assertEqual(exception.returncode, 2)
+        self.assertIn("tee.py", str(exception), "The CommandError does not include the expected command")
+        self.assertEqual(1, exception.returncode, "Unexpected return value from the test pipe")
+        self.assertEqual("TEST_STRING\n", exception.stdout, "Unexpected stdout from the test pipe")
+        self.assertEqual("TEST_STRING\n", exception.stderr, "Unexpected stderr from the test pipe")
 
-    def test_run_command_should_raise_an_exception_when_it_cannot_execute_the_command(self):
-        command = "nonexistent_command"
+    def test_run_command_should_raise_an_exception_when_the_command_fails(self):
+        tee_script = self.__create_tee_script(return_code=1)
 
+        self.__it_should_raise_an_exception_when_the_command_fails(
+            lambda: shellutil.run_command(tee_script, input="TEST_STRING\n"))
+
+    def test_run_pipe_should_raise_an_exception_when_the_last_command_fails(self):
+        tee_script = self.__create_tee_script(return_code=1)
+
+        self.__it_should_raise_an_exception_when_the_command_fails(
+            lambda: shellutil.run_pipe([["echo", "-n", "TEST_STRING\n"], [tee_script]]))
+
+    def __it_should_raise_an_exception_when_it_cannot_execute_the_command(self, action):
         with self.assertRaises(Exception) as context_manager:
-            shellutil.run_command(command)
+            action()
 
         exception = context_manager.exception
         self.assertIn("No such file or directory", str(exception))
 
-    @patch("azurelinuxagent.common.utils.shellutil.logger", autospec=True)
-    def test_run_command_it_should_not_log_by_default(self, mock_logger):
+    def test_run_command_should_raise_an_exception_when_it_cannot_execute_the_command(self):
+        self.__it_should_raise_an_exception_when_it_cannot_execute_the_command(
+            lambda: shellutil.run_command("nonexistent_command"))
+
+    def test_run_pipe_should_raise_an_exception_when_it_cannot_execute_the_pipe(self):
+        self.__it_should_raise_an_exception_when_it_cannot_execute_the_command(
+            lambda: shellutil.run_pipe([["ls", "-ld", "."], ["nonexistent_command"], ["wc", "-l"]]))
 
-        def assert_no_message_logged(command):
+    def __it_should_not_log_by_default(self, action):
+        with patch("azurelinuxagent.common.utils.shellutil.logger", autospec=True) as mock_logger:
             try:
-                shellutil.run_command(command)
-            except:  # pylint: disable=bare-except
+                action()
+            except Exception:
                 pass
 
             self.assertEqual(mock_logger.info.call_count, 0)
@@ -169,50 +228,139 @@ class RunCommandTestCase(AgentTestCase):
             self.assertEqual(mock_logger.warn.call_count, 0)
             self.assertEqual(mock_logger.error.call_count, 0)
 
-            assert_no_message_logged(["ls", "nonexistent_file"])
-            assert_no_message_logged("nonexistent_command")
+    def test_run_command_it_should_not_log_by_default(self):
+        self.__it_should_not_log_by_default(
+            lambda: shellutil.run_command(["ls", "nonexistent_file"]))  # Raises a CommandError
 
-    def test_run_command_it_should_log_an_error_when_log_error_is_set(self):
-        command = ["ls", "-d", "/etc", "nonexistent_file"]
+        self.__it_should_not_log_by_default(
+            lambda: shellutil.run_command("nonexistent_command"))  # Raises an OSError
 
+    def test_run_pipe_it_should_not_log_by_default(self):
+        self.__it_should_not_log_by_default(
+            lambda: shellutil.run_pipe([["date"], [self.__create_tee_script(return_code=1)]]))  # Raises a CommandError
+
+        self.__it_should_not_log_by_default(
+            lambda: shellutil.run_pipe([["date"], ["nonexistent_command"]]))  # Raises an OSError
+
+    def __it_should_log_an_error_when_log_error_is_set(self, action, command):
         with patch("azurelinuxagent.common.utils.shellutil.logger.error") as mock_log_error:
             try:
-                shellutil.run_command(command, log_error=True)
-            except:  # pylint: disable=bare-except
+                action()
+            except Exception:
                 pass
 
             self.assertEqual(mock_log_error.call_count, 1)
 
-            args, kwargs = mock_log_error.call_args  # pylint: disable=unused-variable
-            self.assertIn("ls -d /etc nonexistent_file", args, msg="The command was not logged")
-            self.assertIn(2, args, msg="The command's return code was not logged")
-            self.assertIn("/etc\n", args, msg="The command's stdout was not logged")
-            self.assertTrue(any("No such file or directory" in str(a) for a in args), msg="The command's stderr was not logged")
+            args, _ = mock_log_error.call_args
+            self.assertTrue(any(command in str(a) for a in args), "The command was not logged")
+            self.assertTrue(any("2" in str(a) for a in args), "The command's return code was not logged")  # errno 2: No such file or directory
 
-        command = "nonexistent_command"
+    def test_run_command_should_log_an_error_when_log_error_is_set(self):
+        self.__it_should_log_an_error_when_log_error_is_set(
+            lambda: shellutil.run_command(["ls", "file-does-not-exist"], log_error=True),  # Raises a CommandError
+            command="ls")
 
-        with patch("azurelinuxagent.common.utils.shellutil.logger.error") as mock_log_error:
-            try:
-                shellutil.run_command(command, log_error=True)
-            except:  # pylint: disable=bare-except
-                pass
+        self.__it_should_log_an_error_when_log_error_is_set(
+            lambda: shellutil.run_command("command-does-not-exist", log_error=True),  # Raises a CommandError
+            command="command-does-not-exist")
 
-            self.assertEqual(mock_log_error.call_count, 1)
+    def test_run_command_should_raise_when_both_the_input_and_stdin_parameters_are_specified(self):
+        with tempfile.TemporaryFile() as input_file:
+            with self.assertRaises(ValueError):
+                shellutil.run_command(["cat"], input='0123456789ABCDEF', stdin=input_file)
+
+    def test_run_command_should_read_the_command_input_from_the_input_parameter_when_it_is_a_string(self):
+        command_input = 'TEST STRING'
+        output = shellutil.run_command(["cat"], input=command_input)
+        self.assertEqual(output, command_input, "The command did not process its input correctly; the output should match the input")
+
+    def test_run_command_should_read_stdin_from_the_input_parameter_when_it_is_a_sequence_of_bytes(self):
+        command_input = 'TEST BYTES'
+        output = shellutil.run_command(["cat"], input=command_input)
+        self.assertEqual(output, command_input, "The command did not process its input correctly; the output should match the input")
+
+    def __it_should_read_the_command_input_from_the_stdin_parameter(self, action):
+        command_input = 'TEST STRING\n'
+        with tempfile.TemporaryFile() as input_file:
+            input_file.write(command_input.encode())
+            input_file.seek(0)
+
+            output = action(stdin=input_file)
+
+            self.assertEqual(output, command_input, "The command did not process its input correctly; the output should match the input")
+
+    def test_run_command_should_read_the_command_input_from_the_stdin_parameter(self):
+        self.__it_should_read_the_command_input_from_the_stdin_parameter(
+            lambda stdin: shellutil.run_command(["cat"], stdin=stdin))
+
+    def test_run_pipe_should_read_the_command_input_from_the_stdin_parameter(self):
+        self.__it_should_read_the_command_input_from_the_stdin_parameter(
+            lambda stdin: shellutil.run_pipe([["cat"], ["sort"]], stdin=stdin))
+
+    def __it_should_write_the_command_output_to_the_stdout_parameter(self, action):
+        with tempfile.TemporaryFile() as output_file:
+            captured_output = action(stdout=output_file)
+
+            output_file.seek(0)
+            command_output = ustr(output_file.read(), encoding='utf-8', errors='backslashreplace')
+
+            self.assertEqual(command_output, "TEST STRING\n", "The command did not produce the correct output; the output should match the input")
+            self.assertEqual("", captured_output, "No output should have been captured since it was redirected to a file. Output: [{0}]".format(captured_output))
+
+    def test_run_command_should_write_the_command_output_to_the_stdout_parameter(self):
+        self.__it_should_write_the_command_output_to_the_stdout_parameter(
+            lambda stdout: shellutil.run_command(["echo", "TEST STRING"], stdout=stdout))
+
+    def test_run_pipe_should_write_the_command_output_to_the_stdout_parameter(self):
+        self.__it_should_write_the_command_output_to_the_stdout_parameter(
+            lambda stdout: shellutil.run_pipe([["echo", "TEST STRING"], ["sort"]], stdout=stdout))
+
+    def __it_should_write_the_command_error_output_to_the_stderr_parameter(self, action):
+        with tempfile.TemporaryFile() as output_file:
+            action(stderr=output_file)
+
+            output_file.seek(0)
+            command_error_output = ustr(output_file.read(), encoding='utf-8', errors="backslashreplace")
+
+            self.assertEqual("TEST STRING\n", command_error_output, "stderr was not redirected to the output file correctly")
+
+    def test_run_command_should_write_the_command_error_output_to_the_stderr_parameter(self):
+        self.__it_should_write_the_command_error_output_to_the_stderr_parameter(
+            lambda stderr: shellutil.run_command(self.__create_tee_script(), input="TEST STRING\n", stderr=stderr))
+
+    def test_run_pipe_should_write_the_command_error_output_to_the_stderr_parameter(self):
+        self.__it_should_write_the_command_error_output_to_the_stderr_parameter(
+            lambda stderr: shellutil.run_pipe([["echo", "TEST STRING"], [self.__create_tee_script()]], stderr=stderr))
+
+    def test_run_pipe_should_capture_the_stderr_of_all_the_commands_in_the_pipe(self):
+        with self.assertRaises(shellutil.CommandError) as context_manager:
+            shellutil.run_pipe([
+                ["echo", "TEST STRING"],
+                [self.__create_tee_script()],
+                [self.__create_tee_script()],
+                [self.__create_tee_script(return_code=1)]])
+
+        self.assertEqual("TEST STRING\n" * 3, context_manager.exception.stderr, "Expected 3 copies of the test string since there are 3 commands in the pipe")
+
+    def test_run_command_should_return_a_string_by_default(self):
+        output = shellutil.run_command(self.__create_tee_script(), input="TEST STRING")
+
+        self.assertTrue(isinstance(output, ustr), "The return value should be a string. Got: '{0}'".format(type(output)))
+
+    def test_run_pipe_should_return_a_string_by_default(self):
+        output = shellutil.run_pipe([["echo", "TEST STRING"], [self.__create_tee_script()]])
+
+        self.assertTrue(isinstance(output, ustr), "The return value should be a string. Got: '{0}'".format(type(output)))
+
+    def test_run_command_should_return_a_bytes_object_when_encode_output_is_false(self):
+        output = shellutil.run_command(self.__create_tee_script(), input="TEST STRING", encode_output=False)
 
-            args, kwargs = mock_log_error.call_args
-            self.assertIn(command, args, msg="The command was not logged")
-            self.assertTrue(any("No such file or directory" in str(a) for a in args), msg="The command's stderr was not logged")
+        self.assertTrue(isinstance(output, bytes), "The return value should be a bytes object. Got: '{0}'".format(type(output)))
 
-    def test_run_command_it_should_read_from_stdin_if_cmd_input_is_set(self):
-        import random
-        command = ["cat"]
-        random_hash = ''.join(random.choice('0123456789ABCDEF') for _ in range(16))
-        try:
-            output = shellutil.run_command(command, cmd_input=random_hash)
-        except:  # pylint: disable=bare-except
-            self.fail("No exception should've been thrown when trying to read from stdin in run_command")
+    def test_run_pipe_should_return_a_bytes_object_when_encode_output_is_false(self):
+        output = shellutil.run_pipe([["echo", "TEST STRING"], [self.__create_tee_script()]], encode_output=False)
 
-        self.assertEqual(output, random_hash, "We're reading from stdin and printing it shell, output should match")
+        self.assertTrue(isinstance(output, bytes), "The return value should be a bytes object. Got: '{0}'".format(type(output)))
 
 
 if __name__ == '__main__':