[device/celestica] Mitigation for command injection vulnerability

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

@@ -1,8 +1,9 @@
 import os
+import ast
 import imp
 import yaml
 import subprocess
-
+from shlex import split
 from sonic_py_common import device_info
 
 
@@ -47,7 +48,7 @@ def run_command(self, command):
         output = ""
         try:
             p = subprocess.Popen(
-                command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+                split(command), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
             raw_data, err = p.communicate()
             if p.returncode == 0:
                 status, output = True, raw_data.strip()
@@ -67,7 +68,7 @@ def _clean_input(self, input, config):
             cleaned_input = input_translator.get(input)
 
         elif type(input_translator) is str:
-            cleaned_input = eval(input_translator.format(input))
+            cleaned_input = ast.literal_eval(input_translator.format(input))
 
         return cleaned_input
 
@@ -77,9 +78,9 @@ def _clean_output(self, index, output, config):
         if type(output_translator) is dict:
             output = output_translator.get(output)
         elif type(output_translator) is str:
-            output = eval(output_translator.format(output))
+            output = ast.literal_eval(output_translator.format(output))
         elif type(output_translator) is list:
-            output = eval(output_translator[index].format(output))
+            output = ast.literal_eval(output_translator[index].format(output))
 
         return output
 
@@ -195,7 +196,8 @@ def write_txt_file(self, file_path, value):
         return True
 
     def is_host(self):
-        return os.system(self.HOST_CHK_CMD) == 0
+        command = split(self.HOST_CHK_CMD)
+        return subprocess.run(command).returncode == 0
 
     def load_json_file(self, path):
         """

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

@@ -10,8 +10,8 @@
 try:
     import os.path
     import shutil
-    import shlex
     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")
@@ -40,7 +40,7 @@ def __run_command(self, command):
         # Run bash command and print output to stdout
         try:
             process = subprocess.Popen(
-                shlex.split(command), universal_newlines=True, stdout=subprocess.PIPE)
+                split(command), universal_newlines=True, stdout=subprocess.PIPE)
             while True:
                 output = process.stdout.readline()
                 if output == '' and process.poll() is not None:
@@ -65,7 +65,7 @@ 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)
+            split(cmd), universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
         raw_data, err = p.communicate()
         if err is not '':
             return None
@@ -76,11 +76,11 @@ def __get_cpld_version(self):
         cpld_version = dict()
         with open(SMC_CPLD_PATH, 'r') as fd:
             smc_cpld_version = fd.read()
-        smc_cpld_version = 'None' if smc_cpld_version is 'None' else "{}.{}".format(
+        smc_cpld_version = 'None' if smc_cpld_version == 'None' else "{}.{}".format(
             int(smc_cpld_version[2], 16), int(smc_cpld_version[3], 16))
 
         mmc_cpld_version = self.get_register_value(MMC_CPLD_ADDR)
-        mmc_cpld_version = 'None' if mmc_cpld_version is 'None' else "{}.{}".format(
+        mmc_cpld_version = 'None' if mmc_cpld_version == 'None' else "{}.{}".format(
             int(mmc_cpld_version[2], 16), int(mmc_cpld_version[3], 16))
 
         cpld_version["SMC_CPLD"] = smc_cpld_version

device/celestica/x86_64-cel_silverstone-r0/sonic_platform/psu.py

@@ -33,20 +33,15 @@
 PSU_LED_GREEN_CMD = "0x01"
 PSU_LED_AMBER_CMD = "0x02"
 
-PSU1_VOUT_SS_ID = "0x36"
-PSU1_COUT_SS_ID = "0x37"
-PSU1_POUT_SS_ID = "0x38"
-PSU1_STATUS_REG = "0x39"
-
-PSU2_VOUT_SS_ID = "0x40"
-PSU2_COUT_SS_ID = "0x41"
-PSU2_POUT_SS_ID = "0x42"
-PSU2_STATUS_REG = "0x2f"
-
 PSU1_FRU_ID = 3
 
 SS_READ_OFFSET = 0
 
+PSU_VOUT_SS_ID = {"PSU1_VOUT_SS_ID": "0x36", "PSU2_VOUT_SS_ID": "0x40"}
+PSU_COUT_SS_ID = {"PSU1_COUT_SS_ID": "0x37", "PSU2_COUT_SS_ID": "0x41"}
+PSU_POUT_SS_ID = {"PSU1_POUT_SS_ID": "0x38", "PSU2_POUT_SS_ID": "0x42"}
+PSU_STATUS_REG = {"PSU1_STATUS_REG": "0x39", "PSU2_STATUS_REG": "0x2f"}
+
 
 class Psu(PsuBase):
     """Platform-specific Psu class"""
@@ -71,7 +66,7 @@ def get_voltage(self):
             e.g. 12.1
         """
         psu_voltage = 0.0
-        psu_vout_key = globals()['PSU{}_VOUT_SS_ID'.format(self.index+1)]
+        psu_vout_key = PSU_VOUT_SS_ID['PSU{}_VOUT_SS_ID'.format(self.index+1)]
         status, raw_ss_read = self._api_helper.ipmi_raw(
             IPMI_SENSOR_NETFN, IPMI_SS_READ_CMD.format(psu_vout_key))
         ss_read = raw_ss_read.split()[SS_READ_OFFSET]
@@ -87,7 +82,7 @@ def get_current(self):
             A float number, the electric current in amperes, e.g 15.4
         """
         psu_current = 0.0
-        psu_cout_key = globals()['PSU{}_COUT_SS_ID'.format(self.index+1)]
+        psu_cout_key = PSU_COUT_SS_ID['PSU{}_COUT_SS_ID'.format(self.index+1)]
         status, raw_ss_read = self._api_helper.ipmi_raw(
             IPMI_SENSOR_NETFN, IPMI_SS_READ_CMD.format(psu_cout_key))
         ss_read = raw_ss_read.split()[SS_READ_OFFSET]
@@ -103,7 +98,7 @@ def get_power(self):
             A float number, the power in watts, e.g. 302.6
         """
         psu_power = 0.0
-        psu_pout_key = globals()['PSU{}_POUT_SS_ID'.format(self.index+1)]
+        psu_pout_key = PSU_POUT_SS_ID['PSU{}_POUT_SS_ID'.format(self.index+1)]
         status, raw_ss_read = self._api_helper.ipmi_raw(
             IPMI_SENSOR_NETFN, IPMI_SS_READ_CMD.format(psu_pout_key))
         ss_read = raw_ss_read.split()[SS_READ_OFFSET]
@@ -176,7 +171,7 @@ def get_presence(self):
             bool: True if PSU is present, False if not
         """
         psu_presence = False
-        psu_pstatus_key = globals()['PSU{}_STATUS_REG'.format(self.index+1)]
+        psu_pstatus_key = PSU_STATUS_REG['PSU{}_STATUS_REG'.format(self.index+1)]
         status, raw_status_read = self._api_helper.ipmi_raw(
             IPMI_SENSOR_NETFN, IPMI_SS_READ_CMD.format(psu_pstatus_key))
         status_byte = self.find_value(raw_status_read)
@@ -228,7 +223,7 @@ def get_status(self):
             A boolean value, True if device is operating properly, False if not
         """
         psu_status = False
-        psu_pstatus_key = globals()['PSU{}_STATUS_REG'.format(self.index+1)]
+        psu_pstatus_key = PSU_STATUS_REG['PSU{}_STATUS_REG'.format(self.index+1)]
         status, raw_status_read = self._api_helper.ipmi_raw(
             IPMI_SENSOR_NETFN, IPMI_SS_READ_CMD.format(psu_pstatus_key))
         status_byte = self.find_value(raw_status_read)