[device/quanta] Mitigation for security vulnerability (#11867)

Signed-off-by: maipbui <[email protected]>
Dependency: [https://github.com/sonic-net/sonic-buildimage/pull/12065](https://github.com/sonic-net/sonic-buildimage/pull/12065)
#### Why I did it
`shell=True` is dangerous because this call will spawn the command using a shell process
`os` - not secure against maliciously constructed input and dangerous if used to evaluate dynamic content.
#### How I did it
`os` - use with `subprocess`
Use `shell=False` with shell features
- redirection: [https://stackoverflow.com/questions/4965159/how-to-redirect-output-with-subprocess-in-python/6482200#6482200?newreg=53afb91b3ebd47c5930be627fcdf2930](https://stackoverflow.com/questions/4965159/how-to-redirect-output-with-subprocess-in-python/6482200#6482200?newreg=53afb91b3ebd47c5930be627fcdf2930)
- `|` operator: [https://docs.python.org/2/library/subprocess.html#replacing-shell-pipeline](https://docs.python.org/2/library/subprocess.html#replacing-shell-pipeline)

Author: maipbui

device/quanta/x86_64-quanta_ix1b_rglbmc-r0/plugins/psuutil.py

@@ -7,6 +7,7 @@
 import os.path
 import subprocess
 import logging
+from sonic_py_common.general import check_output_pipe
 
 try:
     from sonic_psu.psu_base import PsuBase
@@ -22,11 +23,13 @@ def show_log(txt):
     return
 
 
-def exec_cmd(cmd, show):
+def exec_cmd(cmd_args, out_file, show):
+    cmd = ' '.join(cmd_args) + ' > ' + out_file
     logging.info('Run :'+cmd)
     try:
-        output = subprocess.check_output(cmd, shell=True, universal_newlines=True)
-        show_log(cmd + "output:"+str(output))
+        with open(out_file, 'w') as f:
+            output = subprocess.check_output(cmd_args, stdout=f, universal_newlines=True)
+            show_log(cmd + "output:"+str(output))
     except subprocess.CalledProcessError as e:
         logging.info("Failed :"+cmd)
         if show:
@@ -40,12 +43,13 @@ def my_log(txt):
     return
 
 
-def log_os_system(cmd, show):
+def log_os_system(cmd1_args, cmd2_args, show):
+    cmd = ' '.join(cmd1_args) + ' | ' + ' '.join(cmd2_args)
     logging.info('Run :'+cmd)
     status = 1
     output = ""
     try:
-        output = subprocess.check_output(cmd, shell=True, universal_newlines=True)
+        output = check_output_pipe(cmd1_args, cmd2_args)
         my_log(cmd + "output:"+str(output))
     except subprocess.CalledProcessError as e:
         logging.info('Failed :'+cmd)
@@ -55,28 +59,28 @@ def log_os_system(cmd, show):
 
 
 def gpio16_exist():
-    ls = log_os_system("ls /sys/class/gpio/ | grep gpio16", 0)
+    ls = log_os_system(["ls", "/sys/class/gpio/"], ["grep", "gpio16"], 0)
     logging.info('mods:'+ls)
     if len(ls) == 0:
         return False
 
 
 def gpio17_exist():
-    ls = log_os_system("ls /sys/class/gpio/ | grep gpio17", 0)
+    ls = log_os_system(["ls", "/sys/class/gpio/"], ["grep", "gpio17"], 0)
     logging.info('mods:'+ls)
     if len(ls) == 0:
         return False
 
 
 def gpio19_exist():
-    ls = log_os_system("ls /sys/class/gpio/ | grep gpio19", 0)
+    ls = log_os_system(["ls", "/sys/class/gpio/"], ["grep", "gpio19"], 0)
     logging.info('mods:'+ls)
     if len(ls) == 0:
         return False
 
 
 def gpio20_exist():
-    ls = log_os_system("ls /sys/class/gpio/ | grep gpio20", 0)
+    ls = log_os_system(["ls", "/sys/class/gpio/"], ["grep", "gpio20"], 0)
     logging.info('mods:'+ls)
     if len(ls) == 0:
         return False
@@ -95,20 +99,20 @@ def __init__(self):
         PsuBase.__init__(self)
 
         if gpio16_exist() == False:
-            output = exec_cmd("echo 16 > /sys/class/gpio/export ", 1)
-            output = exec_cmd("echo in > /sys/class/gpio/gpio16/direction ", 1)
+            output = exec_cmd(["echo", "16"], "/sys/class/gpio/export", 1)
+            output = exec_cmd(["echo", "in"], "/sys/class/gpio/gpio16/direction", 1)
 
         if gpio17_exist() == False:
-            output = exec_cmd("echo 17 > /sys/class/gpio/export ", 1)
-            output = exec_cmd("echo in > /sys/class/gpio/gpio17/direction ", 1)
+            output = exec_cmd(["echo", "17"], "/sys/class/gpio/export", 1)
+            output = exec_cmd(["echo", "in"], "/sys/class/gpio/gpio17/direction", 1)
 
         if gpio19_exist() == False:
-            output = exec_cmd("echo 19 > /sys/class/gpio/export ", 1)
-            output = exec_cmd("echo in > /sys/class/gpio/gpio19/direction ", 1)
+            output = exec_cmd(["echo", "19"], "/sys/class/gpio/export", 1)
+            output = exec_cmd(["echo", "in"], "/sys/class/gpio/gpio19/direction", 1)
 
         if gpio20_exist() == False:
-            output = exec_cmd("echo 20 > /sys/class/gpio/export ", 1)
-            output = exec_cmd("echo in > /sys/class/gpio/gpio20/direction ", 1)
+            output = exec_cmd(["echo", "20"], "/sys/class/gpio/export", 1)
+            output = exec_cmd(["echo", "in"], "/sys/class/gpio/gpio20/direction", 1)
 
     # Get sysfs attribute
     def get_attr_value(self, attr_path):

platform/broadcom/sonic-platform-modules-quanta/ix1b-32x/utils/quanta_ix1b_util.py

@@ -27,22 +27,19 @@
     clean       : uninstall drivers and remove related sysfs nodes    
 """
 
-import os
 import commands
 import sys, getopt
 import logging
-import re
 import time
-from collections import namedtuple
 
 DEBUG = False
 args = []
 FORCE = 0
 i2c_prefix = '/sys/bus/i2c/devices/'
 
 if DEBUG == True:
-    print sys.argv[0]
-    print 'ARGV      :', sys.argv[1:]
+    print(sys.argv[0])
+    print('ARGV      :', sys.argv[1:])
 
 def main():
     global DEBUG
@@ -56,10 +53,10 @@ def main():
                                                        'debug',
                                                        'force',
                                                           ])
-    if DEBUG == True:
-        print options
-        print args
-        print len(sys.argv)
+    if DEBUG is True:
+        print(options)
+        print(args)
+        print(len(sys.argv))
 
     for opt, arg in options:
         if opt in ('-h', '--help'):
@@ -83,12 +80,12 @@ def main():
     return 0
 
 def show_help():
-    print __doc__ % {'scriptName' : sys.argv[0].split("/")[-1]}
+    print(__doc__ % {'scriptName' : sys.argv[0].split("/")[-1]})
     sys.exit(0)
 
 def show_log(txt):
-    if DEBUG == True:
-        print "[IX1B-32X]" + txt
+    if DEBUG is True:
+        print("[IX1B-32X]" + txt)
 
     return
 
@@ -165,7 +162,7 @@ def system_install():
         status, output = exec_cmd("modprobe " + drivers[i], 1)
 
     if status:
-        print output
+        print(output)
         if FORCE == 0:
             return status
 
@@ -174,7 +171,7 @@ def system_install():
         status, output = exec_cmd(instantiate[i], 1)
 
     if status:
-        print output
+        print(output)
         if FORCE == 0:
             return status
 
@@ -184,7 +181,9 @@ def system_install():
     #QSFP for 1~32 port
     for port_number in range(1, 33):
         bus_number = port_number + 31
-        os.system("echo %d >/sys/bus/i2c/devices/%d-0050/port_name" % (port_number, bus_number))
+        file = "/sys/bus/i2c/devices/%d-0050/port_name" % bus_number
+        with open(file, 'w') as f:
+            f.write(str(port_number) + '\n')
 
     #Set system LED to green
     status, output = exec_cmd("echo 1 > /sys/class/leds/sysled_green/brightness", 1)
@@ -199,14 +198,14 @@ def system_ready():
 
 def install():
     if not device_found():
-        print "No device, installing...."
+        print("No device, installing....")
         status = system_install()
 
         if status:
             if FORCE == 0:
                 return status
     else:
-        print " ix1b driver already installed...."
+        print(" ix1b driver already installed....")
 
     return
 
@@ -215,10 +214,10 @@ def uninstall():
 
     #uninstall drivers
     for i in range(len(drivers) - 1, -1, -1):
-       status, output = exec_cmd("rmmod " + drivers[i], 1)
+        status, output = exec_cmd("rmmod " + drivers[i], 1)
 
     if status:
-        print output
+        print(output)
         if FORCE == 0:
             return status
 

platform/broadcom/sonic-platform-modules-quanta/ix7-32x/sonic_platform/component.py

@@ -14,6 +14,7 @@
     import subprocess
     from sonic_platform_base.component_base import ComponentBase
     from collections import namedtuple
+    from sonic_py_common.general import getstatusoutput_noshell_pipe
 except ImportError as e:
     raise ImportError(str(e) + "- required module not found")
 
@@ -47,7 +48,7 @@ def _get_command_result(cmdline):
         try:
             proc = subprocess.Popen(cmdline,
                                     stdout=subprocess.PIPE,
-                                    shell=True, stderr=subprocess.STDOUT,
+                                    stderr=subprocess.STDOUT,
                                     universal_newlines=True)
             stdout = proc.communicate()[0]
             rc = proc.wait()
@@ -60,12 +61,24 @@ def _get_command_result(cmdline):
 
         return result
 
+    @staticmethod
+    def _get_command_result_pipe(cmd1, cmd2):
+        try:
+            rc, result = getstatusoutput_noshell_pipe(cmd1, cmd2)
+            if rc != [0, 0]:
+                raise RuntimeError("Failed to execute command {} {}, return code {}, {}".format(cmd1, cmd2, rc, result))
+
+        except OSError as e:
+            raise RuntimeError("Failed to execute command {} {} due to {}".format(cmd1, cmd2, repr(e)))
+
+        return result
+
 
 class ComponentBIOS(Component):
     COMPONENT_NAME = 'BIOS'
     COMPONENT_DESCRIPTION = 'BIOS - Basic Input/Output System'
 
-    BIOS_QUERY_VERSION_COMMAND = "dmidecode -s bios-version"
+    BIOS_QUERY_VERSION_COMMAND = ["dmidecode", "-s", "bios-version"]
 
     def __init__(self):
         super(ComponentBIOS, self).__init__()
@@ -90,7 +103,8 @@ def get_firmware_version(self):
 class ComponentBMC(Component):
     COMPONENT_NAME = 'BMC'
     COMPONENT_DESCRIPTION = 'BMC - Board Management Controller'
-    BMC_QUERY_VERSION_COMMAND = "ipmitool mc info | grep 'Firmware Revision'"
+    BMC_QUERY_VERSION_COMMAND1 = ["ipmitool", "mc", "info"]
+    BMC_QUERY_VERSION_COMMAND2 = ["grep", 'Firmware Revision']
 
     def __init__(self):
         super(ComponentBMC, self).__init__()
@@ -105,7 +119,7 @@ def get_firmware_version(self):
         Returns:
             A string containing the firmware version of the component
         """
-        bmc_ver = self._get_command_result(self.BMC_QUERY_VERSION_COMMAND)
+        bmc_ver = self._get_command_result_pipe(self.BMC_QUERY_VERSION_COMMAND1, self.BMC_QUERY_VERSION_COMMAND2)
         if not bmc_ver:
             return 'ERR'
         else:
@@ -159,7 +173,7 @@ def get_firmware_version(self):
         Returns:
             A string containing the firmware version of the component
         """
-        res = self._get_command_result("ipmitool raw 0x32 0xff 0x02 {}".format(self.cplds[self.index].cmd_index))
+        res = self._get_command_result(["ipmitool", "raw", "0x32", "0xff", "0x02", str(self.cplds[self.index].cmd_index)])
         if not res:
             return 'ERR'
         else:
@@ -179,7 +193,8 @@ def get_component_list(cls):
 class ComponentPCIE(Component):
     COMPONENT_NAME = 'PCIe'
     COMPONENT_DESCRIPTION = 'ASIC PCIe Firmware'
-    PCIE_QUERY_VERSION_COMMAND = "bcmcmd 'pciephy fw version' | grep 'FW version'"
+    PCIE_QUERY_VERSION_COMMAND1 = ["bcmcmd", 'pciephy fw version']
+    PCIE_QUERY_VERSION_COMMAND2 = ["grep", 'FW version']
 
     def __init__(self):
         super(ComponentPCIE, self).__init__()
@@ -194,7 +209,7 @@ def get_firmware_version(self):
         Returns:
             A string containing the firmware version of the component
         """
-        version = self._get_command_result(self.PCIE_QUERY_VERSION_COMMAND)
+        version = self._get_command_result_pipe(self.PCIE_QUERY_VERSION_COMMAND1, self.PCIE_QUERY_VERSION_COMMAND2)
         if not version:
             return 'ERR'
         else:

platform/broadcom/sonic-platform-modules-quanta/ix7-32x/sonic_platform/sfp.py

@@ -8,17 +8,17 @@
 #
 #############################################################################
 
-import os
 import time
+import subprocess
 from ctypes import create_string_buffer
 
 try:
-     from sonic_platform_base.sfp_base import SfpBase
-     from sonic_platform_base.sonic_sfp.sff8472 import sff8472InterfaceId
-     from sonic_platform_base.sonic_sfp.sff8472 import sff8472Dom
-     from sonic_platform_base.sonic_sfp.sff8436 import sff8436InterfaceId
-     from sonic_platform_base.sonic_sfp.sff8436 import sff8436Dom
-     from sonic_platform_base.sonic_sfp.sfputilhelper import SfpUtilHelper
+    from sonic_platform_base.sfp_base import SfpBase
+    from sonic_platform_base.sonic_sfp.sff8472 import sff8472InterfaceId
+    from sonic_platform_base.sonic_sfp.sff8472 import sff8472Dom
+    from sonic_platform_base.sonic_sfp.sff8436 import sff8436InterfaceId
+    from sonic_platform_base.sonic_sfp.sff8436 import sff8436Dom
+    from sonic_platform_base.sonic_sfp.sfputilhelper import SfpUtilHelper
 except ImportError as e:
     raise ImportError(str(e) + "- required module not found")
 
@@ -163,7 +163,7 @@ class Sfp(SfpBase):
     # Path to QSFP sysfs
     PLATFORM_ROOT_PATH = "/usr/share/sonic/device"
     PMON_HWSKU_PATH = "/usr/share/sonic/hwsku"
-    HOST_CHK_CMD = "docker > /dev/null 2>&1"
+    HOST_CHK_CMD = ["docker"]
 
     PLATFORM = "x86_64-quanta_ix7_rglbmc-r0"
     HWSKU  = "Quanta-IX7-32X"
@@ -259,7 +259,7 @@ def _convert_string_to_num(self, value_str):
             return 'N/A'
 
     def __is_host(self):
-        return os.system(self.HOST_CHK_CMD) == 0
+        return subprocess.call(self.HOST_CHK_CMD) == 0
 
     def __get_path_to_port_config_file(self):
         platform_path = "/".join([self.PLATFORM_ROOT_PATH, self.PLATFORM])

platform/broadcom/sonic-platform-modules-quanta/ix7-32x/utils/quanta_ix7_util.py

@@ -27,7 +27,6 @@
     clean       : uninstall drivers and remove related sysfs nodes
 """
 
-import os
 import subprocess
 import sys, getopt
 import logging
@@ -54,7 +53,7 @@ def main():
                                                        'debug',
                                                        'force',
                                                           ])
-    if DEBUG == True:
+    if DEBUG is True:
         print(options)
         print(args)
         print(len(sys.argv))
@@ -84,7 +83,7 @@ def show_help():
     sys.exit(0)
 
 def show_log(txt):
-    if DEBUG == True:
+    if DEBUG is True:
         print("[IX7-32X]" + txt)
     return
 
@@ -204,7 +203,9 @@ def system_install():
     #QSFP for 1~32 port
     for port_number in range(1, 33):
         bus_number = port_number + 16
-        os.system("echo %d >/sys/bus/i2c/devices/%d-0050/port_name" % (port_number, bus_number))
+        file = "/sys/bus/i2c/devices/%d-0050/port_name" % bus_number
+        with open(file, 'w') as f:
+            f.write(str(port_number) + '\n')
     return