[utilities_common] replace shell=True (sonic-net#2718)

#### What I did
`subprocess()` - when using with `shell=True` is dangerous. Using subprocess function without a static string can lead to command injection.
#### How I did it
remove shell=True in utilities_common.cli.run_command() function.
`subprocess()` - use `shell=False` instead, use list of strings Ref: [https://semgrep.dev/docs/cheat-sheets/python-command-injection/#mitigation](https://semgrep.dev/docs/cheat-sheets/python-command-injection/#mitigation)
#### How to verify it
Pass UT
Manual test
Signed-off-by: Mai Bui <[email protected]>

Author: maipbui

clear/main.py

@@ -492,10 +492,10 @@ def flowcnt_route(ctx, namespace):
     """Clear all route flow counters"""
     exit_if_route_flow_counter_not_support()
     if ctx.invoked_subcommand is None:
-        command = "flow_counters_stat -c -t route"
+        command = ['flow_counters_stat', '-c', '-t', 'route']
         # None namespace means default namespace
         if namespace is not None:
-            command += " -n {}".format(namespace)
+            command += ['-n', str(namespace)]
         clicommon.run_command(command)
 
 
@@ -506,12 +506,12 @@ def flowcnt_route(ctx, namespace):
 @click.argument('prefix-pattern', required=True)
 def pattern(prefix_pattern, vrf, namespace):
     """Clear route flow counters by pattern"""
-    command = "flow_counters_stat -c -t route --prefix_pattern {}".format(prefix_pattern)
+    command = ['flow_counters_stat', '-c', '-t', 'route', '--prefix_pattern', str(prefix_pattern)]
     if vrf:
-        command += ' --vrf {}'.format(vrf)
+        command += ['--vrf', str(vrf)]
     # None namespace means default namespace
     if namespace is not None:
-        command += " -n {}".format(namespace)
+        command += ['-n', str(namespace)]
     clicommon.run_command(command)
 
 
@@ -522,12 +522,12 @@ def pattern(prefix_pattern, vrf, namespace):
 @click.argument('prefix', required=True)
 def route(prefix, vrf, namespace):
     """Clear route flow counters by prefix"""
-    command = "flow_counters_stat -c -t route --prefix {}".format(prefix)
+    command = ['flow_counters_stat', '-c', '-t', 'route', '--prefix', str(prefix)]
     if vrf:
-        command += ' --vrf {}'.format(vrf)
+        command += ['--vrf', str(vrf)]
     # None namespace means default namespace
     if namespace is not None:
-        command += " -n {}".format(namespace)
+        command += ['-n', str(namespace)]
     clicommon.run_command(command)
 
 

config/main.py

@@ -52,7 +52,7 @@
 SNIFFER_CONF_FILE = '/etc/supervisor/conf.d/mlnx_sniffer.conf'
 SNIFFER_CONF_FILE_IN_CONTAINER = CONTAINER_NAME + ':' + SNIFFER_CONF_FILE
 # Command to restart swss service
-COMMAND_RESTART_SWSS = 'systemctl restart swss.service'
+COMMAND_RESTART_SWSS = ['systemctl', 'restart', 'swss.service']
 
 
 # Global logger instance
@@ -99,12 +99,12 @@ def env_variable_delete(delete_line):
 
 
 def conf_file_copy(src, dest):
-    command = 'docker cp ' + src + ' ' + dest
+    command = ['docker', 'cp', str(src), str(dest)]
     clicommon.run_command(command)
 
 
 def conf_file_receive():
-    command = "docker exec {} bash -c 'touch {}'".format(CONTAINER_NAME, SNIFFER_CONF_FILE)
+    command = ['docker', 'exec', str(CONTAINER_NAME), 'bash', '-c', 'touch ' + str(SNIFFER_CONF_FILE)]
     clicommon.run_command(command)
     conf_file_copy(SNIFFER_CONF_FILE_IN_CONTAINER, TMP_SNIFFER_CONF_FILE)
 
@@ -134,7 +134,7 @@ def sniffer_env_variable_set(enable, env_variable_name, env_variable_string=""):
     if not ignore:
         config_file_send()
 
-    command = 'rm -rf {}'.format(TMP_SNIFFER_CONF_FILE)
+    command = ['rm', '-rf', str(TMP_SNIFFER_CONF_FILE)]
     clicommon.run_command(command)
 
     return ignore

config/plugins/mlnx.py

@@ -410,8 +410,8 @@ def add(db, server_ip_address, source, port, vrf):
 
     try:
         add_entry(db.cfgdb, table, key, data)
-        clicommon.run_command("systemctl reset-failed rsyslog-config rsyslog", display_cmd=True)
-        clicommon.run_command("systemctl restart rsyslog-config", display_cmd=True)
+        clicommon.run_command(['systemctl', 'reset-failed', 'rsyslog-config', 'rsyslog'], display_cmd=True)
+        clicommon.run_command(['systemctl', 'restart', 'rsyslog-config'], display_cmd=True)
         log.log_notice("Added remote syslog logging: server={},source={},port={},vrf={}".format(
             server_ip_address,
             data.get(SYSLOG_SOURCE, "N/A"),
@@ -442,8 +442,8 @@ def delete(db, server_ip_address):
 
     try:
         del_entry(db.cfgdb, table, key)
-        clicommon.run_command("systemctl reset-failed rsyslog-config rsyslog", display_cmd=True)
-        clicommon.run_command("systemctl restart rsyslog-config", display_cmd=True)
+        clicommon.run_command(['systemctl', 'reset-failed', 'rsyslog-config', 'rsyslog'], display_cmd=True)
+        clicommon.run_command(['systemctl', 'restart', 'rsyslog-config'], display_cmd=True)
         log.log_notice("Removed remote syslog logging: server={}".format(server_ip_address))
     except Exception as e:
         log.log_error("Failed to remove remote syslog logging: {}".format(str(e)))

config/syslog.py

@@ -139,31 +139,30 @@ def del_vlan(db, vid, no_restart_dhcp_relay):
 
 
 def restart_ndppd():
-    verify_swss_running_cmd = "docker container inspect -f '{{.State.Status}}' swss"
-    docker_exec_cmd = "docker exec -i swss {}"
-    ndppd_status_cmd= "supervisorctl status ndppd"
-    ndppd_conf_copy_cmd = "cp /usr/share/sonic/templates/ndppd.conf /etc/supervisor/conf.d/"
-    supervisor_update_cmd = "supervisorctl update"
-    ndppd_config_gen_cmd = "sonic-cfggen -d -t /usr/share/sonic/templates/ndppd.conf.j2,/etc/ndppd.conf"
-    ndppd_restart_cmd = "supervisorctl restart ndppd"
+    verify_swss_running_cmd = ['docker', 'container', 'inspect', '-f', '{{.State.Status}}', 'swss']
+    docker_exec_cmd = ['docker', 'exec', '-i', 'swss']
+    ndppd_config_gen_cmd = ['sonic-cfggen', '-d', '-t', '/usr/share/sonic/templates/ndppd.conf.j2,/etc/ndppd.conf']
+    ndppd_restart_cmd =['supervisorctl', 'restart', 'ndppd']
+    ndppd_status_cmd= ["supervisorctl", "status", "ndppd"]
+    ndppd_conf_copy_cmd = ['cp', '/usr/share/sonic/templates/ndppd.conf', '/etc/supervisor/conf.d/']
+    supervisor_update_cmd = ['supervisorctl', 'update']
 
     output, _ = clicommon.run_command(verify_swss_running_cmd, return_cmd=True)
 
     if output and output.strip() != "running":
         click.echo(click.style('SWSS container is not running, changes will take effect the next time the SWSS container starts', fg='red'),)
         return
 
-    _, rc = clicommon.run_command(docker_exec_cmd.format(ndppd_status_cmd), ignore_error=True, return_cmd=True)
+    _, rc = clicommon.run_command(docker_exec_cmd + ndppd_status_cmd, ignore_error=True, return_cmd=True)
 
     if rc != 0:
-        clicommon.run_command(docker_exec_cmd.format(ndppd_conf_copy_cmd))
-        clicommon.run_command(docker_exec_cmd.format(supervisor_update_cmd), return_cmd=True)
+        clicommon.run_command(docker_exec_cmd + ndppd_conf_copy_cmd)
+        clicommon.run_command(docker_exec_cmd + supervisor_update_cmd, return_cmd=True)
 
     click.echo("Starting ndppd service")
-    clicommon.run_command(docker_exec_cmd.format(ndppd_config_gen_cmd))
+    clicommon.run_command(docker_exec_cmd + ndppd_config_gen_cmd)
     sleep(3)
-    clicommon.run_command(docker_exec_cmd.format(ndppd_restart_cmd), return_cmd=True)
-
+    clicommon.run_command(docker_exec_cmd + ndppd_restart_cmd, return_cmd=True)
 
 @vlan.command('proxy_arp')
 @click.argument('vid', metavar='<vid>', required=True, type=int)

config/vlan.py

@@ -50,12 +50,12 @@ def check_bootchart_installed():
 
 def get_enabled_status():
     """ Get systemd-bootchart status """
-    output, _ = clicommon.run_command("systemctl is-enabled systemd-bootchart", return_cmd=True)
+    output, _ = clicommon.run_command(['systemctl', 'is-enabled', 'systemd-bootchart'], return_cmd=True)
     return output
 
 def get_active_status():
     """ Get systemd-bootchart status """
-    output, _ = clicommon.run_command("systemctl is-active systemd-bootchart", return_cmd=True)
+    output, _ = clicommon.run_command(['systemctl', 'is-active', 'systemd-bootchart'], return_cmd=True)
     return output
 
 def get_output_files():
@@ -75,14 +75,14 @@ def cli():
 @root_privileges_required
 def enable():
     """ Enable bootchart """
-    clicommon.run_command("systemctl enable systemd-bootchart", display_cmd=True)
+    clicommon.run_command(['systemctl', 'enable', 'systemd-bootchart'], display_cmd=True)
 
 
 @cli.command()
 @root_privileges_required
 def disable():
     """ Disable bootchart """
-    clicommon.run_command("systemctl disable systemd-bootchart", display_cmd=True)
+    clicommon.run_command(['systemctl', 'disable', 'systemd-bootchart'], display_cmd=True)
 
 
 @cli.command()

scripts/sonic-bootchart

@@ -19,13 +19,13 @@ def acl():
 @click.option('--verbose', is_flag=True, help="Enable verbose output")
 def rule(table_name, rule_id, verbose):
     """Show existing ACL rules"""
-    cmd = "acl-loader show rule"
+    cmd = ['acl-loader', 'show', 'rule']
 
     if table_name is not None:
-        cmd += " {}".format(table_name)
+        cmd += [str(table_name)]
 
     if rule_id is not None:
-        cmd += " {}".format(rule_id)
+        cmd += [str(rule_id)]
 
     clicommon.run_command(cmd, display_cmd=verbose)
 
@@ -36,9 +36,9 @@ def rule(table_name, rule_id, verbose):
 @click.option('--verbose', is_flag=True, help="Enable verbose output")
 def table(table_name, verbose):
     """Show existing ACL tables"""
-    cmd = "acl-loader show table"
+    cmd = ['acl-loader', 'show', 'table']
 
     if table_name is not None:
-        cmd += " {}".format(table_name)
+        cmd += [str(table_name)]
 
     clicommon.run_command(cmd, display_cmd=verbose)

show/acl.py

@@ -117,13 +117,13 @@ def midplane_status(chassis_module_name):
 def system_ports(systemportname, namespace, verbose):
     """Show VOQ system ports information"""
 
-    cmd = "voqutil -c system_ports"
+    cmd = ['voqutil', '-c', 'system_ports']
 
     if systemportname is not None:
-        cmd += " -i \"{}\"".format(systemportname)
+        cmd += ['-i', str(systemportname)]
 
     if namespace is not None:
-        cmd += " -n {}".format(namespace)
+        cmd += ['-n', str(namespace)]
 
     clicommon.run_command(cmd, display_cmd=verbose)
 
@@ -134,13 +134,13 @@ def system_ports(systemportname, namespace, verbose):
 def system_neighbors(asicname, ipaddress, verbose):
     """Show VOQ system neighbors information"""
 
-    cmd = "voqutil -c system_neighbors"
+    cmd = ['voqutil', '-c', 'system_neighbors']
 
     if ipaddress is not None:
-        cmd += " -a {}".format(ipaddress)
+        cmd += ['-a', str(ipaddress)]
 
     if asicname is not None:
-        cmd += " -n {}".format(asicname)
+        cmd += ['-n', str(asicname)]
 
     clicommon.run_command(cmd, display_cmd=verbose)
 
@@ -152,15 +152,15 @@ def system_neighbors(asicname, ipaddress, verbose):
 def system_lags(systemlagname, asicname, linecardname, verbose):
     """Show VOQ system lags information"""
 
-    cmd = "voqutil -c system_lags"
+    cmd = ['voqutil', '-c', 'system_lags']
 
     if systemlagname is not None:
-        cmd += " -s \"{}\"".format(systemlagname)
+        cmd += ['-s', str(systemlagname)]
 
     if asicname is not None:
-        cmd += " -n {}".format(asicname)
+        cmd += ['-n', str(asicname)]
 
     if linecardname is not None:
-        cmd += " -l \"{}\"".format(linecardname)
+        cmd += ['-l', str(linecardname)]
 
     clicommon.run_command(cmd, display_cmd=verbose)

show/chassis_modules.py

@@ -18,10 +18,10 @@ def dropcounters():
 @click.option('--verbose', is_flag=True, help="Enable verbose output")
 def configuration(group, verbose):
     """Show current drop counter configuration"""
-    cmd = "dropconfig -c show_config"
+    cmd = ['dropconfig', '-c', 'show_config']
 
     if group:
-        cmd += " -g '{}'".format(group)
+        cmd += ['-g', str(group)]
 
     clicommon.run_command(cmd, display_cmd=verbose)
 
@@ -31,7 +31,7 @@ def configuration(group, verbose):
 @click.option('--verbose', is_flag=True, help="Enable verbose output")
 def capabilities(verbose):
     """Show device drop counter capabilities"""
-    cmd = "dropconfig -c show_capabilities"
+    cmd = ['dropconfig', '-c', 'show_capabilities']
 
     clicommon.run_command(cmd, display_cmd=verbose)
 
@@ -43,12 +43,12 @@ def capabilities(verbose):
 @click.option('--verbose', is_flag=True, help="Enable verbose output")
 def counts(group, counter_type, verbose):
     """Show drop counts"""
-    cmd = "dropstat -c show"
+    cmd = ['dropstat', '-c', 'show']
 
     if group:
-        cmd += " -g '{}'".format(group)
+        cmd += ['-g', str(group)]
 
     if counter_type:
-        cmd += " -t '{}'".format(counter_type)
+        cmd += ['-t', str(counter_type)]
 
     clicommon.run_command(cmd, display_cmd=verbose)

show/dropcounters.py

@@ -18,30 +18,30 @@ def counters():
 @click.option('-e', '--errors', is_flag=True)
 def reachability(namespace, errors):
     """Show fabric reachability"""
-    cmd = "fabricstat -r"
+    cmd = ['fabricstat', '-r']
     if namespace is not None:
-        cmd += " -n {}".format(namespace)
+        cmd += ['-n', str(namespace)]
     if errors:
-        cmd += " -e"
+        cmd += ["-e"]
     clicommon.run_command(cmd)
 
 @counters.command()
 @multi_asic_util.multi_asic_click_option_namespace
 @click.option('-e', '--errors', is_flag=True)
 def port(namespace, errors):
     """Show fabric port stat"""
-    cmd = "fabricstat"
+    cmd = ["fabricstat"]
     if namespace is not None:
-        cmd += " -n {}".format(namespace)
+        cmd += ['-n', str(namespace)]
     if errors:
-        cmd += " -e"
+        cmd += ["-e"]
     clicommon.run_command(cmd)
 
 @counters.command()
 @multi_asic_util.multi_asic_click_option_namespace
 def queue(namespace):
     """Show fabric queue stat"""
-    cmd = "fabricstat -q"
+    cmd = ['fabricstat', '-q']
     if namespace is not None:
-        cmd += " -n {}".format(namespace)
+        cmd += ['-n', str(namespace)]
     clicommon.run_command(cmd)

show/fabric.py

@@ -21,9 +21,9 @@ def flowcnt_trap():
 @click.option('--namespace', '-n', 'namespace', default=None, type=click.Choice(multi_asic_util.multi_asic_ns_choices()), show_default=True, help='Namespace name or all')
 def stats(verbose, namespace):
     """Show trap flow counter statistic"""
-    cmd = "flow_counters_stat -t trap"
+    cmd = ['flow_counters_stat', '-t', 'trap']
     if namespace is not None:
-        cmd += " -n {}".format(namespace)
+        cmd += ['-n', str(namespace)]
     clicommon.run_command(cmd, display_cmd=verbose)
 
 #
@@ -57,9 +57,9 @@ def config(db):
 def stats(ctx, verbose, namespace):
     """Show statistics of all route flow counters"""
     if ctx.invoked_subcommand is None:
-        command = "flow_counters_stat -t route"
+        command = ['flow_counters_stat', '-t', 'route']
         if namespace is not None:
-            command += " -n {}".format(namespace)
+            command += ['-n', str(namespace)]
         clicommon.run_command(command, display_cmd=verbose)
 
 
@@ -70,11 +70,11 @@ def stats(ctx, verbose, namespace):
 @click.argument('prefix-pattern', required=True)
 def pattern(prefix_pattern, vrf, verbose, namespace):
     """Show statistics of route flow counters by pattern"""
-    command = "flow_counters_stat -t route --prefix_pattern \"{}\"".format(prefix_pattern)
+    command = ['flow_counters_stat', '-t', 'route', '--prefix_pattern', str(prefix_pattern)]
     if vrf:
-        command += ' --vrf {}'.format(vrf)
+        command += ['--vrf', str(vrf)]
     if namespace is not None:
-        command += " -n {}".format(namespace)
+        command += ['-n', str(namespace)]
     clicommon.run_command(command, display_cmd=verbose)
 
 
@@ -85,9 +85,9 @@ def pattern(prefix_pattern, vrf, verbose, namespace):
 @click.argument('prefix', required=True)
 def route(prefix, vrf, verbose, namespace):
     """Show statistics of route flow counters by prefix"""
-    command = "flow_counters_stat -t route --prefix {}".format(prefix)
+    command = ['flow_counters_stat', '-t', 'route', '--prefix', str(prefix)]
     if vrf:
-        command += ' --vrf {}'.format(vrf)
+        command += ['--vrf', str(vrf)]
     if namespace is not None:
-        command += " -n {}".format(namespace)
+        command += ['-n', str(namespace)]
     clicommon.run_command(command, display_cmd=verbose)