Remove os.system and subprocess shell=True

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

Author: maipbui

device/celestica/x86_64-cel_e1031-r0/sonic_platform/common.py

@@ -3,7 +3,6 @@
 import imp
 import yaml
 import subprocess
-from shlex import split
 from sonic_py_common import device_info
 
 
@@ -25,6 +24,7 @@ class Common:
 
     SET_METHOD_IPMI = 'ipmitool'
     NULL_VAL = 'N/A'
+    HOST_CHK_CMD = "docker"
     REF_KEY = '$ref:'
 
     def __init__(self, conf=None):
@@ -46,8 +46,7 @@ def run_command(self, command):
         status = False
         output = ""
         try:
-            p = subprocess.Popen(
-                split(command), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
             raw_data, err = p.communicate()
             if p.returncode == 0:
                 status, output = True, raw_data.strip()
@@ -82,14 +81,7 @@ def _clean_output(self, index, output, config):
             output = ast.literal_eval(output_translator[index].format(output))
 
         return output
-
-    def _ipmi_get(self, index, config):
-        argument = config.get('argument')
-        cmd = config['command'].format(
-            config['argument'][index]) if argument else config['command']
-        status, output = self.run_command(cmd)
-        return output if status else None
-
+    
     def _sysfs_read(self, index, config):
         sysfs_path = config.get('sysfs_path')
         argument = config.get('argument', '')
@@ -132,10 +124,6 @@ def _sysfs_write(self, index, config, input):
             return False, output
         return True, output
 
-    def _ipmi_set(self, index, config, input):
-        arg = config['argument'][index].format(input)
-        return self.run_command(config['command'].format(arg))
-
     def _hex_ver_decode(self, hver, num_of_bits, num_of_points):
         ver_list = []
         c_bit = 0
@@ -159,14 +147,16 @@ def _get_class(self, config):
         return class_
 
     def get_reg(self, path, reg_addr):
-        cmd = "echo {1} > {0}; cat {0}".format(path, reg_addr)
-        status, output = self.run_command(cmd)
-        return output if status else None
+        with open(path, 'w') as file:
+            file.write(reg_addr)
+        with open(path, 'r') as file:
+            output = file.readline().strip()
+        return output
 
     def set_reg(self, path, reg_addr, value):
-        cmd = "echo {0} {1} > {2}".format(reg_addr, value, path)
-        status, output = self.run_command(cmd)
-        return output if status else None
+        with open(path, 'w') as file:
+            file.write("{0} {1}".format(reg_addr, value))
+        return None
 
     def read_txt_file(self, path):
         try:
@@ -195,7 +185,7 @@ def write_txt_file(self, file_path, value):
         return True
 
     def is_host(self):
-        return subprocess.run('docker', stdout=None, stderr=None).returncode == 0
+        return subprocess.run(self.HOST_CHK_CMD, stdout=None, stderr=None).returncode == 0
 
     def load_json_file(self, path):
         """
@@ -221,88 +211,7 @@ def get_config_path(self, config_name):
         """
         return os.path.join(self.DEVICE_PATH, self.platform, self.CONFIG_DIR, config_name) if self.is_host() else os.path.join(self.PMON_PLATFORM_PATH, self.CONFIG_DIR, config_name)
 
-    def get_output(self, index, config, default):
-        """
-        Retrieves the output for each function base on config
-
-        Args:
-            index: An integer containing the index of device.
-            config: A dict object containing the configuration of specified function.
-            default: A string containing the default output of specified function.
-
-        Returns:
-            A string containing the output of specified function in config
-        """
-        output_source = config.get('output_source')
-
-        if output_source == self.OUTPUT_SOURCE_IPMI:
-            output = self._ipmi_get(index, config)
-
-        elif output_source == self.OUTPUT_SOURCE_GIVEN_VALUE:
-            output = config["value"]
-
-        elif output_source == self.OUTPUT_SOURCE_GIVEN_CLASS:
-            output = self._get_class(config)
-
-        elif output_source == self.OUTPUT_SOURCE_GIVEN_LIST:
-            output = config["value_list"][index]
-
-        elif output_source == self.OUTPUT_SOURCE_SYSFS:
-            output = self._sysfs_read(index, config)
-
-        elif output_source == self.OUTPUT_SOURCE_FUNC:
-            func_conf = self._main_conf[config['function'][index]]
-            output = self.get_output(index, func_conf, default)
-
-        elif output_source == self.OUTPUT_SOURCE_GIVEN_TXT_FILE:
-            path = config.get('path')
-            output = self.read_txt_file(path)
-
-        elif output_source == self.OUTPUT_SOURCE_GIVEN_VER_HEX_FILE:
-            path = config.get('path')
-            hex_ver = self.read_txt_file(path)
-            output = self._hex_ver_decode(
-                hex_ver, config['num_of_bits'], config['num_of_points'])
-
-        elif output_source == self.OUTPUT_SOURCE_GIVEN_VER_HEX_ADDR:
-            path = config.get('path')
-            addr = config.get('reg_addr')
-            hex_ver = self.get_reg(path, addr)
-            output = self._hex_ver_decode(
-                hex_ver, config['num_of_bits'], config['num_of_points'])
-
-        else:
-            output = default
-
-        return self._clean_output(index, output, config) or default
-
-    def set_output(self, index, input, config):
-        """
-        Sets the output of specified function on config
-
-        Args:
-            config: A dict object containing the configuration of specified function.
-            index: An integer containing the index of device.
-            input: A string containing the input of specified function.
-
-        Returns:
-            bool: True if set function is successfully, False if not
-        """
-        cleaned_input = self._clean_input(input, config)
-        if not cleaned_input:
-            return False
-
-        set_method = config.get('set_method')
-        if set_method == self.SET_METHOD_IPMI:
-            output = self._ipmi_set(index, config, cleaned_input)[0]
-        elif set_method == self.OUTPUT_SOURCE_SYSFS:
-            output = self._sysfs_write(index, config, cleaned_input)[0]
-        else:
-            output = False
-
-        return output
-
-    def get_event(self, timeout, config, sfp_list):
+        def get_event(self, timeout, config, sfp_list):
         """
         Returns a nested dictionary containing all devices which have
         experienced a change at chassis level

device/celestica/x86_64-cel_e1031-r0/sonic_platform/component.py

@@ -11,7 +11,6 @@
     import os.path
     import shutil
     import subprocess
-    from shlex import split
     from sonic_platform_base.component_base import ComponentBase
 except ImportError as e:
     raise ImportError(str(e) + "- required module not found")
@@ -39,8 +38,7 @@ def __init__(self, component_index):
     def __run_command(self, command):
         # Run bash command and print output to stdout
         try:
-            process = subprocess.Popen(
-                split(command), universal_newlines=True, stdout=subprocess.PIPE)
+            process = subprocess.Popen(command, universal_newlines=True, stdout=subprocess.PIPE)
             while True:
                 output = process.stdout.readline()
                 if output == '' and process.poll() is not None:
@@ -63,12 +61,10 @@ def __get_bios_version(self):
 
     def get_register_value(self, register):
         # Retrieves the cpld register value
-        cmd = "echo {1} > {0}; cat {0}".format(GETREG_PATH, register)
-        p = subprocess.Popen(
-            split(cmd), universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        raw_data, err = p.communicate()
-        if err is not '':
-            return None
+        with open(GETREG_PATH, 'w') as file:
+            file.write(register)
+        with open(GETREG_PATH, 'r') as file:
+            raw_data = file.readline()
         return raw_data.strip()
 
     def __get_cpld_version(self):
@@ -159,7 +155,7 @@ def install_firmware(self, image_path):
             ext = ".vme" if ext == "" else ext
             new_image_path = os.path.join("/tmp", (root.lower() + ext))
             shutil.copy(image_path, new_image_path)
-            install_command = "ispvm %s" % new_image_path
+            install_command = ["ispvm", str(new_image_path)]
         # elif self.name == "BIOS":
         #     install_command = "afulnx_64 %s /p /b /n /x /r" % image_path
         return self.__run_command(install_command)

device/celestica/x86_64-cel_e1031-r0/sonic_platform/thermal_actions.py

@@ -73,6 +73,5 @@ def execute(self, thermal_info_dict):
         thermal_overload_position = Common().read_txt_file(
             thermal_overload_position_path)
 
-        cmd = 'bash /usr/share/sonic/platform/thermal_overload_control.sh {}'.format(
-            thermal_overload_position)
+        cmd = ['bash', '/usr/share/sonic/platform/thermal_overload_control.sh', thermal_overload_position]
         Common().run_command(cmd)

device/celestica/x86_64-cel_seastone-r0/sonic_platform/component.py

@@ -52,12 +52,10 @@ def __get_bios_version(self):
 
     def get_register_value(self, register):
         # Retrieves the cpld register value
-        cmd = "echo {1} > {0}; cat {0}".format(GETREG_PATH, register)
-        p = subprocess.Popen(
-            cmd, shell=True, universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        raw_data, err = p.communicate()
-        if err is not '':
-            return None
+        with open(GETREG_PATH, 'w') as file:
+            file.write(register)
+        with open(GETREG_PATH, 'r') as file:
+            raw_data = file.readline()
         return raw_data.strip()
 
     def __get_cpld_version(self):
@@ -146,11 +144,11 @@ def install_firmware(self, image_path):
             ext = ".vme" if ext == "" else ext
             new_image_path = os.path.join("/tmp", (root.lower() + ext))
             shutil.copy(image_path, new_image_path)
-            install_command = "ispvm %s" % new_image_path
+            install_command = ["ispvm", str(new_image_path)]
         # elif self.name == "BIOS":
         #     install_command = "afulnx_64 %s /p /b /n /x /r" % image_path
 
-        return self.__run_command(install_command)
+        return self_api_helper.run_command(install_command)
 
 
     def update_firmware(self, image_path):

device/celestica/x86_64-cel_seastone-r0/sonic_platform/helper.py

@@ -5,7 +5,7 @@
 
 from sonic_py_common import device_info
 
-HOST_CHK_CMD = "docker > /dev/null 2>&1"
+HOST_CHK_CMD = "docker"
 EMPTY_STRING = ""
 
 
@@ -15,7 +15,7 @@ def __init__(self):
         (self.platform, self.hwsku) = device_info.get_platform_and_hwsku()
 
     def is_host(self):
-        return os.system(HOST_CHK_CMD) == 0
+        return subprocess.run(HOST_CHK_CMD, stdout=None, stderr=None).returncode == 0
 
     def pci_get_value(self, resource, offset):
         status = True
@@ -35,21 +35,14 @@ def run_command(self, cmd):
         result = ""
         try:
             p = subprocess.Popen(
-                cmd, shell=True, universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+                cmd, universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
             raw_data, err = p.communicate()
             if err == '':
                 result = raw_data.strip()
         except:
             status = False
         return status, result
 
-    def run_interactive_command(self, cmd):
-        try:
-            os.system(cmd)
-        except:
-            return False
-        return True
-
     def read_txt_file(self, file_path):
         try:
             with open(file_path, 'r') as fd:
@@ -77,17 +70,19 @@ def write_txt_file(self, file_path, value):
         return True
 
     def get_cpld_reg_value(self, getreg_path, register):
-        cmd = "echo {1} > {0}; cat {0}".format(getreg_path, register)
-        status, result = self.run_command(cmd)
-        return result if status else None
+        with open(getreg_path, 'w') as file:
+            file.write(register)
+        with open(getreg_path, 'r') as file:
+            result = file.readline()
+        return result
 
     def ipmi_raw(self, netfn, cmd):
         status = True
         result = ""
         try:
-            cmd = "ipmitool raw {} {}".format(str(netfn), str(cmd))
+            cmd = ["ipmitool", "raw", str(netfn), str(cmd)]
             p = subprocess.Popen(
-                cmd, shell=True, universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+                cmd, universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
             raw_data, err = p.communicate()
             if err == '':
                 result = raw_data.strip()
@@ -100,29 +95,30 @@ def ipmi_raw(self, netfn, cmd):
     def ipmi_fru_id(self, id, key=None):
         status = True
         result = ""
+        cmd1_args = ["ipmitool", "fru", "print", str(id)]
         try:
-            cmd = "ipmitool fru print {}".format(str(
-                id)) if not key else "ipmitool fru print {0} | grep '{1}' ".format(str(id), str(key))
-
-            p = subprocess.Popen(
-                cmd, shell=True, universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-            raw_data, err = p.communicate()
-            if err == '':
-                result = raw_data.strip()
+            if not key:
+                p = subprocess.Popen(
+                    cmd1_args, universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+                raw_data, err = p.communicate()
             else:
-                status = False
-        except:
+                cmd2_args = ["grep", str(key)]
+                with subprocess.Popen(cmd1_args, universal_newlines=True, stdout=subprocess.PIPE) as p1:
+                    with subprocess.Popen(cmd2_args, universal_newlines=True, stdin=p1.stdout, stdout=subprocess.PIPE) as p2:
+                        raw_data = p2.communicate()[0]
+                        if p2.returncode == 1:
+                            raise subprocess.CalledProcessError(returncode=p2.returncode, cmd=cmd, output=raw_data)
+        except subprocess.CalledProcessError:
             status = False
-        return status, result
+        return status, raw_data.strip()
 
     def ipmi_set_ss_thres(self, id, threshold_key, value):
         status = True
         result = ""
         try:
-            cmd = "ipmitool sensor thresh '{}' {} {}".format(
-                str(id), str(threshold_key), str(value))
+            cmd = ["ipmitool", "sensor", "thresh", str(id), str(threshold_key), str(value)]
             p = subprocess.Popen(
-                cmd, shell=True, universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+                cmd, universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
             raw_data, err = p.communicate()
             if err == '':
                 result = raw_data.strip()

device/celestica/x86_64-cel_seastone-r0/sonic_platform/thermal_actions.py

@@ -73,6 +73,5 @@ def execute(self, thermal_info_dict):
         thermal_overload_position = APIHelper().read_one_line_file(
             thermal_overload_position_path)
 
-        cmd = 'bash /usr/share/sonic/platform/thermal_overload_control.sh {}'.format(
-            thermal_overload_position)
+        cmd = ['bash', '/usr/share/sonic/platform/thermal_overload_control.sh', str(thermal_overload_position)]
         APIHelper().run_command(cmd)

device/celestica/x86_64-cel_seastone-r0/sonic_platform/thermal_manager.py

@@ -5,7 +5,7 @@
 from .thermal_infos import *
 
 class ThermalManager(ThermalManagerBase):
-    FSC_ALGORITHM_CMD = 'service fancontrol {}'
+    FSC_ALGORITHM_CMD = ['service', 'fancontrol', '']
 
     @classmethod
     def start_thermal_control_algorithm(cls):
@@ -43,4 +43,5 @@ def _enable_fancontrol_service(cls, enable):
             bool: True if set success, False if fail.
         """
         cmd = 'start' if enable else 'stop'
-        return APIHelper().run_command(cls.FSC_ALGORITHM_CMD.format(cmd))
+        cls.FSC_ALGORITHM_CMD[2] = cmd
+        return APIHelper().run_command(cls.FSC_ALGORITHM_CMD.format)