[device/dell] Mitigation for security vulnerability (#11875)

Dependency: [PR (#12065)](#12065) needs to merge first. #### Why I did it `commands` module is not protected against malicious input `getstatusoutput` is detected without a static string, uses `shell=True` #### How I did it Eliminate the use of `commands` Use `subprocess.run()`, commands in `subprorcess.run()` are totally static Fix indentation #### How to verify it Tested on DUT [dell_log.txt](https://github.com/sonic-net/sonic-buildimage/files/9561332/dell_log.txt)
sonic-net · Jan 6, 2023 · 06e1a0b · 06e1a0b
1 parent ba5c26a
commit 06e1a0b
Show file tree

Hide file tree

Showing 36 changed files with 722 additions and 683 deletions.
diff --git a/device/dell/x86_64-dellemc_n3248pxe_c3338-r0/plugins/fanutil.py b/device/dell/x86_64-dellemc_n3248pxe_c3338-r0/plugins/fanutil.py
@@ -3,10 +3,10 @@
 # Platform-specific FAN status interface for SONiC
 #
 
-import commands
 import sys
+from sonic_py_common.general import getstatusoutput_noshell
 
-SENSORS_CMD = "docker exec -i pmon /usr/bin/sensors"
+SENSORS_CMD = ["docker", "exec", "-i", "pmon", "/usr/bin/sensors"]
 DOCKER_SENSORS_CMD = "/usr/bin/sensors"
 
 
@@ -33,24 +33,23 @@ def isDockerEnv(self):
             return True
 
     def get_num_fans(self):
-       n3248pxe_MAX_FANTRAYS = 3
-       return n3248pxe_MAX_FANTRAYS
+        n3248pxe_MAX_FANTRAYS = 3
+        return n3248pxe_MAX_FANTRAYS
 
     def get_presence(self, idx):
-       sysfs_path = "/sys/devices/platform/dell-n3248pxe-cpld.0/fan" + self._fan_mapping[idx] + "_prs"
-       return int(open(sysfs_path).read(), 16)
+        sysfs_path = "/sys/devices/platform/dell-n3248pxe-cpld.0/fan" + self._fan_mapping[idx] + "_prs"
+        return int(open(sysfs_path).read(), 16)
 
     def get_direction(self, idx):
-       sysfs_path = "/sys/devices/platform/dell-n3248pxe-cpld.0/fan" + self._fan_mapping[idx] + "_dir"
-       return open(sysfs_path).read()
+        sysfs_path = "/sys/devices/platform/dell-n3248pxe-cpld.0/fan" + self._fan_mapping[idx] + "_dir"
+        return open(sysfs_path).read()
 
     def get_speed(self, idx):
         dockerenv = self.isDockerEnv()
         if not dockerenv:
-            status, cmd_output = commands.getstatusoutput(SENSORS_CMD)
-        else :
-            status, cmd_output = commands.getstatusoutput(DOCKER_SENSORS_CMD)
-
+            status, cmd_output = getstatusoutput_noshell(SENSORS_CMD)
+        else:
+            status, cmd_output = getstatusoutput_noshell(DOCKER_SENSORS_CMD)
         if status:
             print('Failed to execute sensors command')
             sys.exit(0)
@@ -64,9 +63,9 @@ def get_speed(self, idx):
         return 0.0
 
     def get_status(self, idx):
-       sysfs_path = "/sys/devices/platform/dell-n3248pxe-cpld.0/fan" + self._fan_mapping[idx] + "_prs"
-       return int(open(sysfs_path).read(), 16)
+        sysfs_path = "/sys/devices/platform/dell-n3248pxe-cpld.0/fan" + self._fan_mapping[idx] + "_prs"
+        return int(open(sysfs_path).read(), 16)
 
 
     def set_speed(self, idx):
-       return False
+        return False
diff --git a/device/dell/x86_64-dellemc_n3248pxe_c3338-r0/plugins/psuutil.py b/device/dell/x86_64-dellemc_n3248pxe_c3338-r0/plugins/psuutil.py
@@ -3,11 +3,11 @@
 # Platform-specific PSU status interface for SONiC
 #
 
-import commands
 import os
 import sys
+from sonic_py_common.general import getstatusoutput_noshell
 
-SENSORS_CMD = "docker exec -i pmon /usr/bin/sensors"
+SENSORS_CMD = ["docker", "exec", "-i", "pmon", "/usr/bin/sensors"]
 DOCKER_SENSORS_CMD = "/usr/bin/sensors"
 
 try:
@@ -95,10 +95,9 @@ def get_psu_presence(self, index):
     def get_sensor(self):
         dockerenv = self.isDockerEnv()
         if not dockerenv:
-          status, cmd_output = commands.getstatusoutput(SENSORS_CMD)
-        else :
-          status, cmd_output = commands.getstatusoutput(DOCKER_SENSORS_CMD)
-
+            status, cmd_output = getstatusoutput_noshell(SENSORS_CMD)
+        else:
+            status, cmd_output = getstatusoutput_noshell(DOCKER_SENSORS_CMD)
         if status:
             print('Failed to execute sensors command')
             sys.exit(0)

diff --git a/device/dell/x86_64-dellemc_n3248te_c3338-r0/plugins/fanutil.py b/device/dell/x86_64-dellemc_n3248te_c3338-r0/plugins/fanutil.py
@@ -3,10 +3,10 @@
 # Platform-specific FAN status interface for SONiC
 #
 
-import subprocess
 import sys
+from sonic_py_common.general import getstatusoutput_noshell
 
-SENSORS_CMD = "docker exec -i pmon /usr/bin/sensors"
+SENSORS_CMD = ["docker", "exec", "-i", "pmon", "/usr/bin/sensors"]
 DOCKER_SENSORS_CMD = "/usr/bin/sensors"
 
 
@@ -33,24 +33,23 @@ def isDockerEnv(self):
             return True
 
     def get_num_fans(self):
-       N3248TE_MAX_FANTRAYS = 3
-       return N3248TE_MAX_FANTRAYS
+        N3248TE_MAX_FANTRAYS = 3
+        return N3248TE_MAX_FANTRAYS
 
     def get_presence(self, idx):
-       sysfs_path = "/sys/devices/platform/dell-n3248te-cpld.0/fan" + self._fan_mapping[idx] + "_prs"
-       return int(open(sysfs_path).read(), 16)
+        sysfs_path = "/sys/devices/platform/dell-n3248te-cpld.0/fan" + self._fan_mapping[idx] + "_prs"
+        return int(open(sysfs_path).read(), 16)
 
     def get_direction(self, idx):
-       sysfs_path = "/sys/devices/platform/dell-n3248te-cpld.0/fan" + self._fan_mapping[idx] + "_dir"
-       return open(sysfs_path).read()
+        sysfs_path = "/sys/devices/platform/dell-n3248te-cpld.0/fan" + self._fan_mapping[idx] + "_dir"
+        return open(sysfs_path).read()
 
     def get_speed(self, idx):
         dockerenv = self.isDockerEnv()
         if not dockerenv:
-            status, cmd_output = subprocess.getstatusoutput(SENSORS_CMD)
-        else :
-            status, cmd_output = subprocess.getstatusoutput(DOCKER_SENSORS_CMD)
-
+            status, cmd_output = getstatusoutput_noshell(SENSORS_CMD)
+        else:
+            status, cmd_output = getstatusoutput_noshell(DOCKER_SENSORS_CMD)
         if status:
             print('Failed to execute sensors command')
             sys.exit(0)
@@ -64,9 +63,9 @@ def get_speed(self, idx):
         return 0.0
 
     def get_status(self, idx):
-       sysfs_path = "/sys/devices/platform/dell-n3248te-cpld.0/fan" + self._fan_mapping[idx] + "_prs"
-       return int(open(sysfs_path).read(), 16)
+        sysfs_path = "/sys/devices/platform/dell-n3248te-cpld.0/fan" + self._fan_mapping[idx] + "_prs"
+        return int(open(sysfs_path).read(), 16)
 
 
     def set_speed(self, idx):
-       return False
+        return False
diff --git a/device/dell/x86_64-dellemc_n3248te_c3338-r0/plugins/psuutil.py b/device/dell/x86_64-dellemc_n3248te_c3338-r0/plugins/psuutil.py
@@ -3,11 +3,11 @@
 # Platform-specific PSU status interface for SONiC
 #
 
-import commands
 import os
 import sys
+from sonic_py_common.general import getstatusoutput_noshell
 
-SENSORS_CMD = "docker exec -i pmon /usr/bin/sensors"
+SENSORS_CMD = ["docker", "exec", "-i", "pmon", "/usr/bin/sensors"]
 DOCKER_SENSORS_CMD = "/usr/bin/sensors"
 
 try:
@@ -95,10 +95,9 @@ def get_psu_presence(self, index):
     def get_sensor(self):
         dockerenv = self.isDockerEnv()
         if not dockerenv:
-          status, cmd_output = commands.getstatusoutput(SENSORS_CMD)
-        else :
-          status, cmd_output = commands.getstatusoutput(DOCKER_SENSORS_CMD)
-
+            status, cmd_output = getstatusoutput_noshell(SENSORS_CMD)
+        else:
+            status, cmd_output = getstatusoutput_noshell(DOCKER_SENSORS_CMD)
         if status:
             print('Failed to execute sensors command')
             sys.exit(0)

diff --git a/device/dell/x86_64-dellemc_s5212f_c3538-r0/plugins/psuutil.py b/device/dell/x86_64-dellemc_s5212f_c3538-r0/plugins/psuutil.py
@@ -5,11 +5,11 @@
 
 import logging
 import sys
-import subprocess
+from sonic_py_common.general import getstatusoutput_noshell, getstatusoutput_noshell_pipe
 
 S5212F_MAX_PSUS = 2
-IPMI_PSU_DATA = "docker exec -it pmon ipmitool sdr list"
-IPMI_PSU_DATA_DOCKER = "ipmitool sdr list"
+IPMI_PSU_DATA = ["docker", "exec", "-it", "pmon", "ipmitool", "sdr", "list"]
+IPMI_PSU_DATA_DOCKER = ["ipmitool", "sdr", "list"]
 PSU_PRESENCE = "PSU{0}_stat"
 # Use this for older firmware
 # PSU_PRESENCE="PSU{0}_prsnt"
@@ -44,7 +44,7 @@ def get_pmc_register(self, reg_name):
         if dockerenv == True:
             ipmi_cmd = IPMI_PSU_DATA_DOCKER
 
-        status, ipmi_sdr_list = subprocess.getstatusoutput(ipmi_cmd)
+        status, ipmi_sdr_list = getstatusoutput_noshell(ipmi_cmd)
 
         if status:
             logging.error('Failed to execute:' + ipmi_sdr_list)
@@ -91,6 +91,8 @@ def get_psu_presence(self, index):
         :param index: An integer, index of the PSU of which to query status
         :return: Boolean, True if PSU is plugged, False if not
         """
-        cmd_status, psu_status = subprocess.getstatusoutput('ipmitool raw 0x04 0x2d ' + hex(0x30 + index) + " | awk '{print substr($0,9,1)}'")
+        ipmi_cmd = ["ipmitool", "raw", "0x04", "0x2d", hex(0x30 + index)]
+        awk_cmd = ['awk', '{print substr($0,9,1)}']
+        cmd_status, psu_status = getstatusoutput_noshell_pipe(ipmi_cmd, awk_cmd)
         return 1 if psu_status == '1' else 0
 
diff --git a/device/dell/x86_64-dellemc_s5224f_c3538-r0/plugins/psuutil.py b/device/dell/x86_64-dellemc_s5224f_c3538-r0/plugins/psuutil.py
@@ -6,7 +6,7 @@
 
 import logging
 import sys
-import subprocess
+from sonic_py_common.general import getstatusoutput_noshell, getstatusoutput_noshell_pipe
 
 S5224F_MAX_PSUS = 2
 IPMI_PSU_DATA = "docker exec -it pmon ipmitool sdr list"
@@ -45,7 +45,7 @@ def get_pmc_register(self, reg_name):
         if dockerenv == True:
             ipmi_cmd = IPMI_PSU_DATA_DOCKER
 
-        status, ipmi_sdr_list = subprocess.getstatusoutput(ipmi_cmd)
+        status, ipmi_sdr_list = getstatusoutput_noshell(ipmi_cmd)
 
         if status:
             logging.error('Failed to execute:' + ipmi_sdr_list)
@@ -92,6 +92,8 @@ def get_psu_presence(self, index):
         :param index: An integer, index of the PSU of which to query status
         :return: Boolean, True if PSU is plugged, False if not
         """
-        cmd_status, psu_status = subprocess.getstatusoutput('ipmitool raw 0x04 0x2d ' + hex(0x30 + index) + " | awk '{print substr($0,9,1)}'")
+        ipmi_cmd = ["ipmitool", "raw", "0x04", "0x2d", hex(0x30 + index)]
+        awk_cmd = ['awk', '{print substr($0,9,1)}']
+        cmd_status, psu_status = getstatusoutput_noshell_pipe(ipmi_cmd, awk_cmd)
         return 1 if psu_status == '1' else 0
 
diff --git a/device/dell/x86_64-dellemc_s5232f_c3538-r0/plugins/psuutil.py b/device/dell/x86_64-dellemc_s5232f_c3538-r0/plugins/psuutil.py
@@ -4,22 +4,17 @@
 #
 
 
-import os.path
 import logging
 import sys
-
-if sys.version_info[0] < 3:
-    import commands
-else:
-    import subprocess as commands
-
+from sonic_py_common.general import getstatusoutput_noshell_pipe
 
 S5232F_MAX_PSUS = 2
-IPMI_PSU1_DATA = "docker exec -it pmon ipmitool raw 0x04 0x2d 0x31 |  awk '{print substr($0,9,1)}'"
-IPMI_PSU1_DATA_DOCKER = "ipmitool raw 0x04 0x2d 0x31 |  awk '{print substr($0,9,1)}'"
-IPMI_PSU2_DATA = "docker exec -it pmon ipmitool raw 0x04 0x2d 0x32 |  awk '{print substr($0,9,1)}'"
-IPMI_PSU2_DATA_DOCKER = "ipmitool raw 0x04 0x2d 0x32 |  awk '{print substr($0,9,1)}'"
+IPMI_PSU1_DATA = ["docker", "exec", "-it", "pmon", "ipmitool", "raw", "0x04", "0x2d", "0x31"]
+IPMI_PSU1_DATA_DOCKER = ["ipmitool", "raw", "0x04", "0x2d", "0x31"]
+IPMI_PSU2_DATA = ["docker", "exec", "-it", "pmon", "ipmitool", "raw", "0x04", "0x2d", "0x32"]
+IPMI_PSU2_DATA_DOCKER = ["ipmitool", "raw", "0x04", "0x2d", "0x32"]
 PSU_PRESENCE = "PSU{0}_stat"
+awk_cmd = ['awk', '{print substr($0,9,1)}']
 # Use this for older firmware
 # PSU_PRESENCE="PSU{0}_prsnt"
 
@@ -44,23 +39,24 @@ def isDockerEnv(self):
             return False
 
     # Fetch a BMC register
-    def get_pmc_register(self, reg_name):
+    def get_pmc_register(self, index):
 
         status = 1
-        ipmi_cmd_1 = IPMI_PSU1_DATA
-        ipmi_cmd_2 = IPMI_PSU1_DATA
+        ipmi_cmd = ''
         dockerenv = self.isDockerEnv()
         if dockerenv == True:
             if index == 1:
-                status, ipmi_sdr_list = commands.getstatusoutput(IPMI_PSU1_DATA_DOCKER)
+                ipmi_cmd = IPMI_PSU1_DATA_DOCKER
             elif index == 2:
-                status, ipmi_sdr_list = commands.getstatusoutput(IPMI_PSU2_DATA_DOCKER)
+                ipmi_cmd = IPMI_PSU2_DATA_DOCKER
         else:
             if index == 1:
-                status, ipmi_sdr_list = commands.getstatusoutput(IPMI_PSU1_DATA)
+                ipmi_cmd = IPMI_PSU1_DATA
             elif index == 2:
-                status, ipmi_sdr_list = commands.getstatusoutput(IPMI_PSU2_DATA)
-
+                ipmi_cmd = IPMI_PSU2_DATA
+        if ipmi_cmd != '':
+            status, ipmi_sdr_list = getstatusoutput_noshell_pipe(ipmi_cmd, awk_cmd)
+
         if status:
             logging.error('Failed to execute ipmitool')
             sys.exit(0)
@@ -87,22 +83,25 @@ def get_psu_status(self, index):
         """
         # Until psu_status is implemented this is hardcoded temporarily
 
-        psu_status = 'f'
+        psu_status = ''
         ret_status = 1
+        ipmi_cmd = ''
         dockerenv = self.isDockerEnv()
         if dockerenv == True:
             if index == 1:
-                ret_status, ipmi_sdr_list = commands.getstatusoutput(IPMI_PSU1_DATA_DOCKER)
+                ipmi_cmd = IPMI_PSU1_DATA_DOCKER
             elif index == 2:
-                ret_status, ipmi_sdr_list = commands.getstatusoutput(IPMI_PSU2_DATA_DOCKER)
+                ipmi_cmd = IPMI_PSU2_DATA_DOCKER
         else:
             if index == 1:
-                ret_status, ipmi_sdr_list = commands.getstatusoutput(IPMI_PSU1_DATA)
+                ipmi_cmd = IPMI_PSU1_DATA
             elif index == 2:
-                ret_status, ipmi_sdr_list = commands.getstatusoutput(IPMI_PSU2_DATA)
+                ipmi_cmd = IPMI_PSU2_DATA
+        if ipmi_cmd != '':
+            ret_status, ipmi_sdr_list = getstatusoutput_noshell_pipe(ipmi_cmd, awk_cmd)
 
         if ret_status:
-            logging.error('Failed to execute ipmitool : ')
+            logging.error('Failed to execute ipmitool')
             sys.exit(0)
 
         psu_status = ipmi_sdr_list
@@ -117,20 +116,23 @@ def get_psu_presence(self, index):
         """
         psu_status = '0'
         ret_status = 1
+        ipmi_cmd = ''
         dockerenv = self.isDockerEnv()
         if dockerenv == True:
             if index == 1:
-                ret_status, ipmi_sdr_list = commands.getstatusoutput(IPMI_PSU1_DATA_DOCKER)
+                ipmi_cmd = IPMI_PSU1_DATA_DOCKER
             elif index == 2:
-                ret_status, ipmi_sdr_list = commands.getstatusoutput(IPMI_PSU2_DATA_DOCKER)
+                ipmi_cmd = IPMI_PSU2_DATA_DOCKER
         else:
             if index == 1:
-                ret_status, ipmi_sdr_list = commands.getstatusoutput(IPMI_PSU1_DATA)
+                ipmi_cmd = IPMI_PSU1_DATA
             elif index == 2:
-                ret_status, ipmi_sdr_list = commands.getstatusoutput(IPMI_PSU2_DATA)
+                ipmi_cmd = IPMI_PSU2_DATA
+        if ipmi_cmd != '':
+            ret_status, ipmi_sdr_list = getstatusoutput_noshell_pipe(ipmi_cmd, awk_cmd)
 
         if ret_status:
-            logging.error('Failed to execute ipmitool : ')
+            logging.error('Failed to execute ipmitool')
             sys.exit(0)
 
         psu_status = ipmi_sdr_list

diff --git a/device/dell/x86_64-dellemc_s5248f_c3538-r0/plugins/psuutil.py b/device/dell/x86_64-dellemc_s5248f_c3538-r0/plugins/psuutil.py
@@ -4,19 +4,14 @@
 #
 
 
-import os.path
 import logging
 import sys
-
-if sys.version_info[0] < 3:
-    import commands
-else:
-    import subprocess as commands
+from sonic_py_common.general import getstatusoutput_noshell
 
 
 S5248F_MAX_PSUS = 2
-IPMI_PSU_DATA = "docker exec -it pmon ipmitool sdr list"
-IPMI_PSU_DATA_DOCKER = "ipmitool sdr list"
+IPMI_PSU_DATA = ["docker", "exec", "-it", "pmon", "ipmitool", "sdr", "list"]
+IPMI_PSU_DATA_DOCKER = ["ipmitool", "sdr", "list"]
 PSU_PRESENCE = "PSU{0}_stat"
 # Use this for older firmware
 # PSU_PRESENCE="PSU{0}_prsnt"
@@ -53,7 +48,7 @@ def get_pmc_register(self, reg_name):
         if dockerenv == True:
             ipmi_cmd = IPMI_PSU_DATA_DOCKER
 
-        status, ipmi_sdr_list = commands.getstatusoutput(ipmi_cmd)
+        status, ipmi_sdr_list = getstatusoutput_noshell(ipmi_cmd)
 
         if status:
             logging.error('Failed to execute:' + ipmi_sdr_list)