replace shell=True

Signed-off-by: maipbui <[email protected]>

show/platform.py

@@ -148,9 +148,9 @@ def temperature():
 @click.argument('args', nargs=-1, type=click.UNPROCESSED)
 def firmware(args):
     """Show firmware information"""
-    cmd = "sudo fwutil show {}".format(" ".join(args))
+    cmd = ["sudo", "fwutil", "show"] + list(args)
 
     try:
-        subprocess.check_call(cmd, shell=True)
+        subprocess.check_call(cmd)
     except subprocess.CalledProcessError as e:
         sys.exit(e.returncode)

show/plugins/barefoot.py

@@ -4,6 +4,7 @@
 import json
 import subprocess
 from sonic_py_common import device_info
+from sonic_py_common.general import getstatusoutput_noshell_pipe, check_output_pipe
 
 @click.group()
 def barefoot():
@@ -25,8 +26,9 @@ def profile():
 
     # Print current profile
     click.echo('Current profile: ', nl=False)
-    subprocess.run('docker exec -it syncd readlink /opt/bfn/install | sed '
-        r's/install_\\\(.\*\\\)_profile/\\1/', check=True, shell=True)
+    cmd0 = ['docker', 'exec', '-it', 'syncd', 'readlink', '/opt/bfn/install']
+    cmd1 = ['sed', r's/install_\\\(.\*\\\)_profile/\\1/']
+    check_output_pipe(cmd0, cmd1)
 
     # Exclude current and unsupported profiles
     opts = ''
@@ -37,9 +39,10 @@ def profile():
 
     # Print profile list
     click.echo('Available profile(s):')
-    subprocess.run('docker exec -it syncd find /opt/bfn -mindepth 1 '
-        r'-maxdepth 1 -type d -name install_\*_profile ' + opts + '| sed '
-        r's%/opt/bfn/install_\\\(.\*\\\)_profile%\\1%', shell=True)
+    cmd0 = ['docker', 'exec', '-it', 'syncd', 'find', '/opt/bfn', '-mindepth', '1',\
+            r'-maxdepth', '1', '-type', 'd', '-name', 'install_\*_profile', r'{}' % opts]
+    cmd1 = ["sed", r's%/opt/bfn/install_\\\(.\*\\\)_profile%\\1%']
+    getstatusoutput_noshell_pipe(cmd0, cmd1)
 
 def register(cli):
     version_info = device_info.get_sonic_version_info()

show/plugins/mlnx.py

@@ -26,7 +26,8 @@
     import sys
     import subprocess
     import click
-    import xml.etree.ElementTree as ET
+    from shlex import join
+    from lxml import etree as ET
     from sonic_py_common import device_info
 except ImportError as e:
     raise ImportError("%s - required module not found" % str(e))
@@ -45,10 +46,11 @@
 def run_command(command, display_cmd=False, ignore_error=False, print_to_console=True):
     """Run bash command and print output to stdout
     """
-    if display_cmd == True:
-        click.echo(click.style("Running command: ", fg='cyan') + click.style(command, fg='green'))
+    if display_cmd:
+        command_str = join(command)
+        click.echo(click.style("Running command: ", fg='cyan') + click.style(command_str, fg='green'))
 
-    proc = subprocess.Popen(command, shell=True, text=True, stdout=subprocess.PIPE)
+    proc = subprocess.Popen(command, text=True, stdout=subprocess.PIPE)
     (out, err) = proc.communicate()
 
     if len(out) > 0 and print_to_console:
@@ -70,17 +72,17 @@ def mlnx():
 # get current status of sniffer from conf file
 def sniffer_status_get(env_variable_name):
     enabled = False
-    command = "docker exec {} bash -c 'touch {}'".format(CONTAINER_NAME, SNIFFER_CONF_FILE)
+    command = ["docker", "exec", CONTAINER_NAME, "bash", "-c", 'touch {}'.format(SNIFFER_CONF_FILE)]
     run_command(command)
-    command = 'docker cp {} {}'.format(SNIFFER_CONF_FILE_IN_CONTAINER, TMP_SNIFFER_CONF_FILE)
+    command = ['docker', 'cp', SNIFFER_CONF_FILE_IN_CONTAINER, TMP_SNIFFER_CONF_FILE]
     run_command(command)
     conf_file = open(TMP_SNIFFER_CONF_FILE, 'r')
     for env_variable_string in conf_file:
         if env_variable_string.find(env_variable_name) >= 0:
             enabled = True
             break
     conf_file.close()
-    command = 'rm -rf {}'.format(TMP_SNIFFER_CONF_FILE)
+    command = ['rm', '-rf', TMP_SNIFFER_CONF_FILE]
     run_command(command)
     return enabled
 
@@ -97,10 +99,8 @@ def is_issu_status_enabled():
     # Get the SAI XML path from sai.profile
     sai_profile_path = '/{}/sai.profile'.format(HWSKU_PATH)
 
-    DOCKER_CAT_COMMAND = 'docker exec {container_name} cat {path}'
-
-    command = DOCKER_CAT_COMMAND.format(container_name=CONTAINER_NAME, path=sai_profile_path)
-    sai_profile_content, _ = run_command(command, print_to_console=False)
+    DOCKER_CAT_COMMAND = ['docker', 'exec', CONTAINER_NAME, 'cat', sai_profile_path]
+    sai_profile_content, _ = run_command(DOCKER_CAT_COMMAND, print_to_console=False)
 
     sai_profile_kvs = {}
 
@@ -117,8 +117,8 @@ def is_issu_status_enabled():
         sys.exit(1)
 
     # Get ISSU from SAI XML
-    command = DOCKER_CAT_COMMAND.format(container_name=CONTAINER_NAME, path=sai_xml_path)
-    sai_xml_content, _ = run_command(command, print_to_console=False)
+    DOCKER_CAT_COMMAND = ['docker', 'exec', CONTAINER_NAME, 'cat', sai_xml_path]
+    sai_xml_content, _ = run_command(DOCKER_CAT_COMMAND, print_to_console=False)
 
     try:
         root = ET.fromstring(sai_xml_content)

tests/fdbshow_test.py

@@ -450,7 +450,7 @@ def test_show_mac_no_address(self):
     def test_show_mac_no_type(self):
         self.set_mock_variant("6")
 
-        result = self.runner.invoke(show.cli.commands["mac"], ["-t Static"])
+        result = self.runner.invoke(show.cli.commands["mac"], ["-t", "Static"])
         print(result.exit_code)
         print(result.output)
         assert result.exit_code == 0

tests/intfstat_test.py

@@ -168,7 +168,7 @@ def test_clear_single_intfs(self):
         result = runner.invoke(show.cli.commands["interfaces"].commands["counters"].commands["rif"], ["Ethernet20"])
         print(result.output)
         # remove the counters snapshot
-        show.run_command("intfstat -D")
+        show.run_command(["intfstat", "-D"])
         assert 'Last cached time was' in result.output.split('\n')[0]
         assert show_interfaces_counters_rif_clear_single_intf in result.output
 
@@ -180,7 +180,7 @@ def test_clear_single_interface_check_all(self):
         result = runner.invoke(show.cli.commands["interfaces"].commands["counters"].commands["rif"], [])
         print(result.stdout)
         # remove the counters snapshot
-        show.run_command("intfstat -D")
+        show.run_command(["intfstat", "-D"])
         assert 'Last cached time was' in result.output.split('\n')[0]
         assert show_single_interface_check_all_clear in result.output
 
@@ -192,7 +192,7 @@ def test_clear(self):
         result = runner.invoke(show.cli.commands["interfaces"].commands["counters"].commands["rif"], [])
         print(result.stdout)
         # remove the counters snapshot
-        show.run_command("intfstat -D")
+        show.run_command(["intfstat", "-D"])
         assert 'Last cached time was' in result.output.split('\n')[0]
         assert show_interfaces_counters_rif_clear in result.output
 

tests/queue_counter_test.py

@@ -1199,7 +1199,7 @@ def test_queue_counters_port_json(self):
         runner = CliRunner()
         result = runner.invoke(
             show.cli.commands["queue"].commands["counters"],
-            ["Ethernet8 --json"]
+            ["Ethernet8", "--json"]
         )
         assert result.exit_code == 0
         print(result.output)
@@ -1224,7 +1224,7 @@ def test_queue_port_voq_counters(self):
         runner = CliRunner()
         result = runner.invoke(
             show.cli.commands["queue"].commands["counters"],
-            ["Ethernet0 --voq"]
+            ["Ethernet0", "--voq"]
         )
         print(result.output)
         assert result.exit_code == 0

tests/show_test.py

@@ -6,7 +6,7 @@
 from unittest import mock
 from unittest.mock import call, MagicMock, patch
 
-EXPECTED_BASE_COMMAND = 'sudo '
+EXPECTED_BASE_COMMAND = ['sudo']
 
 test_path = os.path.dirname(os.path.abspath(__file__))
 modules_path = os.path.dirname(test_path)
@@ -54,65 +54,98 @@ def teardown_class(cls):
         os.environ["UTILITIES_UNIT_TESTING"] = "0"
 
 @patch('show.main.run_command')
+@patch('show.main.run_command_pipe')
 @pytest.mark.parametrize(
         "cli_arguments,expected",
         [
-            ([], 'cat /var/log/syslog'),
-            (['xcvrd'], "cat /var/log/syslog | grep 'xcvrd'"),
-            (['-l', '10'], 'cat /var/log/syslog | tail -10'),
-            (['-f'], 'tail -F /var/log/syslog'),
+            ([], ['cat', '/var/log/syslog']),
+            (['-f'], ['tail', '-F', '/var/log/syslog']),
         ]
 )
-def test_show_logging_default(run_command, cli_arguments, expected):
+@pytest.mark.parametrize(
+        "cli_arguments1,expected0,expected1",
+        [
+            (['xcvrd'], ['cat', '/var/log/syslog'], ['grep', 'xcvrd']),
+            (['-l', '10'], ['cat', '/var/log/syslog'], ['tail', '-10']),
+        ]
+)
+def test_show_logging_default(run_cmd_pipe, run_cmd, cli_arguments, expected, cli_arguments1, expected0, expected1):
     runner = CliRunner()
     result = runner.invoke(show.cli.commands["logging"], cli_arguments)
-    run_command.assert_called_with(EXPECTED_BASE_COMMAND + expected, display_cmd=False)
+    run_cmd.assert_called_with(EXPECTED_BASE_COMMAND + expected, display_cmd=False)
+    result = runner.invoke(show.cli.commands["logging"], cli_arguments1)
+    run_cmd_pipe.assert_called_with(EXPECTED_BASE_COMMAND + expected0, expected1, display_cmd=False)
 
 @patch('show.main.run_command')
+@patch('show.main.run_command_pipe')
 @patch('os.path.isfile', MagicMock(return_value=True))
 @pytest.mark.parametrize(
         "cli_arguments,expected",
         [
-            ([], 'cat /var/log/syslog.1 /var/log/syslog'),
-            (['xcvrd'], "cat /var/log/syslog.1 /var/log/syslog | grep 'xcvrd'"),
-            (['-l', '10'], 'cat /var/log/syslog.1 /var/log/syslog | tail -10'),
-            (['-f'], 'tail -F /var/log/syslog'),
+            ([], ['cat', '/var/log/syslog.1', '/var/log/syslog']),
+            (['-f'], ['tail', '-F', '/var/log/syslog']),
         ]
 )
-def test_show_logging_syslog_1(run_command, cli_arguments, expected):
+@pytest.mark.parametrize(
+        "cli_arguments1,expected0,expected1",
+        [
+            (['xcvrd'], ['cat', '/var/log/syslog.1', '/var/log/syslog'], ['grep', 'xcvrd']),
+            (['-l', '10'], ['cat', '/var/log/syslog.1', '/var/log/syslog'], ['tail', '-10']),
+        ]
+)
+
+def test_show_logging_syslog_1(run_cmd_pipe, run_cmd, cli_arguments, expected, cli_arguments1, expected0, expected1):
     runner = CliRunner()
     result = runner.invoke(show.cli.commands["logging"], cli_arguments)
-    run_command.assert_called_with(EXPECTED_BASE_COMMAND + expected, display_cmd=False)
+    run_cmd.assert_called_with(EXPECTED_BASE_COMMAND + expected, display_cmd=False)
+    result = runner.invoke(show.cli.commands["logging"], cli_arguments1)
+    run_cmd_pipe.assert_called_with(EXPECTED_BASE_COMMAND + expected0, expected1, display_cmd=False)
 
 @patch('show.main.run_command')
+@patch('show.main.run_command_pipe')
 @patch('os.path.exists', MagicMock(return_value=True))
 @pytest.mark.parametrize(
         "cli_arguments,expected",
         [
-            ([], 'cat /var/log.tmpfs/syslog'),
-            (['xcvrd'], "cat /var/log.tmpfs/syslog | grep 'xcvrd'"),
-            (['-l', '10'], 'cat /var/log.tmpfs/syslog | tail -10'),
-            (['-f'], 'tail -F /var/log.tmpfs/syslog'),
+            ([], ['cat', '/var/log.tmpfs/syslog']),
+            (['-f'], ['tail', '-F', '/var/log.tmpfs/syslog']),
         ]
 )
-def test_show_logging_tmpfs(run_command, cli_arguments, expected):
+@pytest.mark.parametrize(
+        "cli_arguments1,expected0,expected1",
+        [
+            (['xcvrd'], ['cat', '/var/log.tmpfs/syslog'], ['grep', 'xcvrd']),
+            (['-l', '10'], ['cat', '/var/log.tmpfs/syslog'], ['tail', '-10']),
+        ]
+)
+def test_show_logging_tmpfs(run_cmd_pipe, run_cmd, cli_arguments, expected, cli_arguments1, expected0, expected1):
     runner = CliRunner()
     result = runner.invoke(show.cli.commands["logging"], cli_arguments)
-    run_command.assert_called_with(EXPECTED_BASE_COMMAND + expected, display_cmd=False)
+    run_cmd.assert_called_with(EXPECTED_BASE_COMMAND + expected, display_cmd=False)
+    result = runner.invoke(show.cli.commands["logging"], cli_arguments1)
+    run_cmd_pipe.assert_called_with(EXPECTED_BASE_COMMAND + expected0, expected1, display_cmd=False)
 
 @patch('show.main.run_command')
+@patch('show.main.run_command_pipe')
 @patch('os.path.isfile', MagicMock(return_value=True))
 @patch('os.path.exists', MagicMock(return_value=True))
 @pytest.mark.parametrize(
         "cli_arguments,expected",
         [
-            ([], 'cat /var/log.tmpfs/syslog.1 /var/log.tmpfs/syslog'),
-            (['xcvrd'], "cat /var/log.tmpfs/syslog.1 /var/log.tmpfs/syslog | grep 'xcvrd'"),
-            (['-l', '10'], 'cat /var/log.tmpfs/syslog.1 /var/log.tmpfs/syslog | tail -10'),
-            (['-f'], 'tail -F /var/log.tmpfs/syslog'),
+            ([], ['cat', '/var/log.tmpfs/syslog.1', '/var/log.tmpfs/syslog']),
+            (['-f'], ['tail', '-F', '/var/log.tmpfs/syslog']),
+        ]
+)
+@pytest.mark.parametrize(
+        "cli_arguments1,expected0,expected1",
+        [
+            (['xcvrd'], ['cat', '/var/log.tmpfs/syslog.1', '/var/log.tmpfs/syslog'], ['grep', 'xcvrd']),
+            (['-l', '10'], ['cat', '/var/log.tmpfs/syslog.1', '/var/log.tmpfs/syslog'], ['tail', '-10']),
         ]
 )
-def test_show_logging_tmpfs_syslog_1(run_command, cli_arguments, expected):
+def test_show_logging_tmpfs_syslog_1(run_cmd_pipe, run_cmd, cli_arguments, expected, cli_arguments1, expected0, expected1):
     runner = CliRunner()
     result = runner.invoke(show.cli.commands["logging"], cli_arguments)
-    run_command.assert_called_with(EXPECTED_BASE_COMMAND + expected, display_cmd=False)
+    run_cmd.assert_called_with(EXPECTED_BASE_COMMAND + expected, display_cmd=False)
+    result = runner.invoke(show.cli.commands["logging"], cli_arguments1)
+    run_cmd_pipe.assert_called_with(EXPECTED_BASE_COMMAND + expected0, expected1, display_cmd=False)

tests/techsupport_test.py

@@ -3,18 +3,18 @@
 from unittest.mock import patch, Mock
 from click.testing import CliRunner
 
-EXPECTED_BASE_COMMAND = 'sudo '
+EXPECTED_BASE_COMMAND = ['sudo']
 
 @patch("show.main.run_command")
 @pytest.mark.parametrize(
         "cli_arguments,expected",
         [
-            ([], 'generate_dump -v -t 5'),
-            (['--since', '2 days ago'], "generate_dump -v -s '2 days ago' -t 5"),
-            (['-g', '50'], 'timeout --kill-after=300s -s SIGTERM --foreground 50m generate_dump -v -t 5'),
-            (['--allow-process-stop'], 'generate_dump -v -a -t 5'),
-            (['--silent'], 'generate_dump -t 5'),
-            (['--debug-dump', '--redirect-stderr'], 'generate_dump -v -d -t 5 -r'),
+            ([], ['generate_dump', '-v', '-t', '5']),
+            (['--since', '2 days ago'], ['generate_dump', '-v', '-s', '2 days ago', '-t', '5']),
+            (['-g', '50'], ['timeout', '--kill-after=300s', '-s', 'SIGTERM', '--foreground', '50m', 'generate_dump', '-v', '-t', '5']),
+            (['--allow-process-stop'], ['generate_dump', '-v', '-a', '-t', '5']),
+            (['--silent'], ['generate_dump', '-t', '5']),
+            (['--debug-dump', '--redirect-stderr'], ['generate_dump', '-v', '-d', '-t', '5', '-r']),
         ]
 )
 def test_techsupport(run_command, cli_arguments, expected):

tests/tunnelstat_test.py

@@ -83,7 +83,7 @@ def test_clear(self):
         expected = show_vxlan_counters_clear_output
 
         # remove the counters snapshot
-        show.run_command("tunnelstat -D")
+        show.run_command(['tunnelstat', '-D'])
         for line in expected:
             assert line in result.output
 
@@ -97,7 +97,7 @@ def test_clear_interface(self):
         expected = show_vxlan_counters_clear_interface_output
 
         # remove the counters snapshot
-        show.run_command("tunnelstat -D")
+        show.run_command(['tunnelstat', '-D'])
         for line in expected:
             assert line in result.output