[device/celestica] Mitigation for command injection vulnerability (#1…

…1740)

Signed-off-by: maipbui <maibui@microsoft.com>
Dependency: [PR (#12065)](#12065) needs to merge first.
#### Why I did it
1. `eval()` - not secure against maliciously constructed input, can be dangerous if used to evaluate dynamic content. This may be a code injection vulnerability.
2. `subprocess()` - when using with `shell=True` is dangerous. Using subprocess function without a static string can lead to command injection.
3. `os` - not secure against maliciously constructed input and dangerous if used to evaluate dynamic content.
4. `is` operator - string comparison should not be used with reference equality.
5. `globals()` - extremely dangerous because it may allow an attacker to execute arbitrary code on the system
#### How I did it
1. `eval()` - use `literal_eval()`
2. `subprocess()` - use `shell=False` instead. use an array string. Ref: [https://semgrep.dev/docs/cheat-sheets/python-command-injection/#mitigation](https://semgrep.dev/docs/cheat-sheets/python-command-injection/#mitigation)
3. `os` - use with `subprocess`
4. `is` - replace by `==` operator for value equality
5. `globals()` - avoid the use of globals()

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

@@ -24,7 +24,6 @@
 RESET_REGISTER = "0x112"
 HOST_REBOOT_CAUSE_PATH = "/host/reboot-cause/previous-reboot-cause.txt"
 PMON_REBOOT_CAUSE_PATH = "/usr/share/sonic/platform/api_files/reboot-cause/previous-reboot-cause.txt"
-HOST_CHK_CMD = "docker > /dev/null 2>&1"
 STATUS_LED_PATH = "/sys/devices/platform/e1031.smc/master_led"
 
 

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

@@ -1,8 +1,8 @@
 import os
+import ast
 import imp
 import yaml
 import subprocess
-
 from sonic_py_common import device_info
 
 
@@ -24,7 +24,7 @@ class Common:
 
     SET_METHOD_IPMI = 'ipmitool'
     NULL_VAL = 'N/A'
-    HOST_CHK_CMD = "docker > /dev/null 2>&1"
+    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(
-                command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            p = subprocess.Popen(command, universal_newlines=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
             raw_data, err = p.communicate()
             if p.returncode == 0:
                 status, output = True, raw_data.strip()
@@ -67,7 +66,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,19 +76,12 @@ 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
 
-    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 + '\n')
+        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}\n".format(reg_addr, value))
+        return None
 
     def read_txt_file(self, path):
         try:
@@ -195,7 +185,11 @@ def write_txt_file(self, file_path, value):
         return True
 
     def is_host(self):
-        return os.system(self.HOST_CHK_CMD) == 0
+        try:
+            subprocess.call(self.HOST_CHK_CMD, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        except FileNotFoundError:
+            return False
+        return True
 
     def load_json_file(self, path):
         """
@@ -221,87 +215,6 @@ 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):
         """
         Returns a nested dictionary containing all devices which have

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

@@ -10,7 +10,6 @@
 try:
     import os.path
     import shutil
-    import shlex
     import subprocess
     from sonic_platform_base.component_base import ComponentBase
 except ImportError as e:
@@ -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(
-                shlex.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,24 +61,22 @@ 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 + '\n')
+        with open(GETREG_PATH, 'r') as file:
+            raw_data = file.readline()
         return raw_data.strip()
 
     def __get_cpld_version(self):
         # Retrieves the CPLD firmware version
         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
@@ -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_e1031-r0/sonic_platform/thermal_manager.py

@@ -6,7 +6,7 @@
 
 
 class ThermalManager(ThermalManagerBase):
-    FSC_ALGORITHM_CMD = ' supervisorctl {} fancontrol'
+    FSC_ALGORITHM_CMD = ['supervisorctl', '', 'fancontrol']
 
     @classmethod
     def start_thermal_control_algorithm(cls):
@@ -43,5 +43,5 @@ def _enable_fancontrol_service(cls, enable):
         Returns:
             bool: True if set success, False if fail.
         """
-        cmd = 'start' if enable else 'stop'
-        return Common().run_command(cls.FSC_ALGORITHM_CMD.format(cmd))
+        cls.FSC_ALGORITHM_CMD[1] = 'start' if enable else 'stop'
+        return Common().run_command(cls.FSC_ALGORITHM_CMD)

device/celestica/x86_64-cel_midstone-r0/plugins/psuutil.py

@@ -6,8 +6,6 @@
 #
 #############################################################################
 
-import os.path
-import subprocess
 
 try:
     from sonic_psu.psu_base import PsuBase

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

@@ -26,7 +26,6 @@
 REBOOT_CAUSE_FILE = "reboot-cause.txt"
 PREV_REBOOT_CAUSE_FILE = "previous-reboot-cause.txt"
 GETREG_PATH = "/sys/devices/platform/dx010_cpld/getreg"
-HOST_CHK_CMD = "docker > /dev/null 2>&1"
 STATUS_LED_PATH = "/sys/devices/platform/leds_dx010/leds/dx010:green:stat/brightness"
 
 

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

@@ -8,7 +8,6 @@
 
 import os.path
 import shutil
-import subprocess
 
 try:
     from sonic_platform_base.component_base import ComponentBase
@@ -52,12 +51,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 + '\n')
+        with open(GETREG_PATH, 'r') as file:
+            raw_data = file.readline()
         return raw_data.strip()
 
     def __get_cpld_version(self):
@@ -146,11 +143,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):