Remove subprocess with shell=True (#24)

Signed-off-by: maipbui [email protected]
Why I did it
subprocess is used with shell=True, which is very dangerous for shell injection.
How I did it
remove shell=True, use shell=False
How to verify it
Pass UT
Manual test

Author: maipbui

scripts/caclmgrd

@@ -13,6 +13,7 @@ from datetime import datetime
 
 from sonic_py_common import daemon_base
 from swsscommon import swsscommon
+from sonic_py_common.general import getstatusoutput_noshell_pipe
 
 VERSION = '1.0'
 
@@ -29,7 +30,7 @@ class ProcDockerStats(daemon_base.DaemonBase):
         self.state_db.connect("STATE_DB")
 
     def run_command(self, cmd):
-        proc = subprocess.Popen(cmd, shell=True, universal_newlines=True, stdout=subprocess.PIPE)
+        proc = subprocess.Popen(cmd, universal_newlines=True, stdout=subprocess.PIPE)
         (stdout, stderr) = proc.communicate()
         if proc.returncode != 0:
             self.log_error("Error running command '{}'".format(cmd))
@@ -118,7 +119,7 @@ class ProcDockerStats(daemon_base.DaemonBase):
         return dockerdict
 
     def update_dockerstats_command(self):
-        cmd = "docker stats --no-stream -a"
+        cmd = ["docker", "stats", "--no-stream", "-a"]
         data = self.run_command(cmd)
         if not data:
             self.log_error("'{}' returned null output".format(cmd))
@@ -135,7 +136,13 @@ class ProcDockerStats(daemon_base.DaemonBase):
         return True
 
     def update_processstats_command(self):
-        data = self.run_command("ps -eo uid,pid,ppid,%mem,%cpu,stime,tty,time,cmd --sort -%cpu | head -1024")
+        cmd0 = ["ps", "-eo", "uid,pid,ppid,%mem,%cpu,stime,tty,time,cmd", "--sort", "-%cpu"]
+        cmd1 = ["head", "-1024"]
+        cmd = ' | '.join([' '.join(cmd0), ' '.join(cmd1)])
+        exitcode, data = getstatusoutput_noshell_pipe(cmd0, cmd1)
+        if any(exitcode):
+            self.log_error("Error running command '{}'".format(cmd))
+            data = None
         processdata = self.format_process_cmd_output(data)
         value = ""
         # wipe out all data before updating with new values

scripts/hostcfgd

@@ -34,17 +34,18 @@ def test_caclmgrd_bfd(self, test_name, test_data, fs):
 
         MockConfigDb.set_config_db(test_data["config_db"])
 
-        with mock.patch("caclmgrd.subprocess") as mocked_subprocess:
-            popen_mock = mock.Mock()
-            popen_attrs = test_data["popen_attributes"]
-            popen_mock.configure_mock(**popen_attrs)
-            mocked_subprocess.Popen.return_value = popen_mock
-            mocked_subprocess.PIPE = -1
-
-            call_rc = test_data["call_rc"]
-            mocked_subprocess.call.return_value = call_rc
-
-            caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
-            caclmgrd_daemon.allow_bfd_protocol('')
-            mocked_subprocess.Popen.assert_has_calls(test_data["expected_subprocess_calls"], any_order=True)
+        with mock.patch("caclmgrd.ControlPlaneAclManager.run_commands_pipe", return_value='sonic'):
+            with mock.patch("caclmgrd.subprocess") as mocked_subprocess:
+                popen_mock = mock.Mock()
+                popen_attrs = test_data["popen_attributes"]
+                popen_mock.configure_mock(**popen_attrs)
+                mocked_subprocess.Popen.return_value = popen_mock
+                mocked_subprocess.PIPE = -1
+
+                call_rc = test_data["call_rc"]
+                mocked_subprocess.call.return_value = call_rc
+
+                caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
+                caclmgrd_daemon.allow_bfd_protocol('')
+                mocked_subprocess.Popen.assert_has_calls(test_data["expected_subprocess_calls"], any_order=True)
 

scripts/procdockerstatsd

@@ -33,21 +33,22 @@ def test_caclmgrd_dhcp(self, test_name, test_data, fs):
 
         MockConfigDb.set_config_db(test_data["config_db"])
 
-        with mock.patch("caclmgrd.subprocess") as mocked_subprocess:
-            popen_mock = mock.Mock()
-            popen_attrs = test_data["popen_attributes"]
-            popen_mock.configure_mock(**popen_attrs)
-            mocked_subprocess.Popen.return_value = popen_mock
+        with mock.patch("caclmgrd.ControlPlaneAclManager.run_commands_pipe", return_value='sonic'):
+            with mock.patch("caclmgrd.subprocess") as mocked_subprocess:
+                popen_mock = mock.Mock()
+                popen_attrs = test_data["popen_attributes"]
+                popen_mock.configure_mock(**popen_attrs)
+                mocked_subprocess.Popen.return_value = popen_mock
 
-            call_rc = test_data["call_rc"]
-            mocked_subprocess.call.return_value = call_rc
+                call_rc = test_data["call_rc"]
+                mocked_subprocess.call.return_value = call_rc
 
-            mark = test_data["mark"]
+                mark = test_data["mark"]
 
-            caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
-            mux_update = test_data["mux_update"]
+                caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
+                mux_update = test_data["mux_update"]
 
-            for key,data in mux_update:
-                caclmgrd_daemon.update_dhcp_acl(key, '', data, mark)
+                for key,data in mux_update:
+                    caclmgrd_daemon.update_dhcp_acl(key, '', data, mark)
 
-            mocked_subprocess.call.assert_has_calls(test_data["expected_subprocess_calls"], any_order=False)
+                mocked_subprocess.call.assert_has_calls(test_data["expected_subprocess_calls"], any_order=False)

tests/caclmgrd/caclmgrd_bfd_test.py

@@ -41,8 +41,10 @@ def test_caclmgrd_external_client_acl(self, test_name, test_data, fs):
         caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
 
         iptables_rules_ret, _ = caclmgrd_daemon.get_acl_rules_and_translate_to_iptables_commands('')
+        test_data['return'] = [tuple(i) for i in test_data['return']]
+        iptables_rules_ret = [tuple(i) for i in iptables_rules_ret]
         self.assertEqual(set(test_data["return"]).issubset(set(iptables_rules_ret)), True)
-        caclmgrd_daemon.iptables_cmd_ns_prefix['asic0'] = 'ip netns exec asic0'
+        caclmgrd_daemon.iptables_cmd_ns_prefix['asic0'] = ['ip', 'netns', 'exec', 'asic0']
         caclmgrd_daemon.namespace_docker_mgmt_ip['asic0'] = '1.1.1.1'
         caclmgrd_daemon.namespace_mgmt_ip = '2.2.2.2'
         caclmgrd_daemon.namespace_docker_mgmt_ipv6['asic0'] = 'fd::01'

tests/caclmgrd/caclmgrd_dhcp_test.py

@@ -34,17 +34,18 @@ def test_feature_present(self, test_name, test_data, fs):
 
         MockConfigDb.set_config_db(test_data["config_db"])
 
-        with mock.patch("caclmgrd.subprocess") as mocked_subprocess:
-            popen_mock = mock.Mock()
-            popen_attrs = test_data["popen_attributes"]
-            popen_mock.configure_mock(**popen_attrs)
-            mocked_subprocess.Popen.return_value = popen_mock
-            mocked_subprocess.PIPE = -1
-
-            call_rc = test_data["call_rc"]
-            mocked_subprocess.call.return_value = call_rc
-
-            caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
-            caclmgrd_daemon.update_feature_present()
-            self.assertTrue("bgp" in caclmgrd_daemon.feature_present)
-            self.assertEqual(caclmgrd_daemon.feature_present["bgp"], True)
+        with mock.patch("caclmgrd.ControlPlaneAclManager.run_commands_pipe", return_value='sonic'):
+            with mock.patch("caclmgrd.subprocess") as mocked_subprocess:
+                popen_mock = mock.Mock()
+                popen_attrs = test_data["popen_attributes"]
+                popen_mock.configure_mock(**popen_attrs)
+                mocked_subprocess.Popen.return_value = popen_mock
+                mocked_subprocess.PIPE = -1
+
+                call_rc = test_data["call_rc"]
+                mocked_subprocess.call.return_value = call_rc
+
+                caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
+                caclmgrd_daemon.update_feature_present()
+                self.assertTrue("bgp" in caclmgrd_daemon.feature_present)
+                self.assertEqual(caclmgrd_daemon.feature_present["bgp"], True)

tests/caclmgrd/caclmgrd_external_client_acl_test.py

@@ -36,19 +36,20 @@ def test_caclmgrd_soc(self, test_name, test_data, fs):
 
         MockConfigDb.set_config_db(test_data["config_db"])
 
-        with mock.patch("caclmgrd.subprocess") as mocked_subprocess:
-            popen_mock = mock.Mock()
-            popen_attrs = test_data["popen_attributes"]
-            popen_mock.configure_mock(**popen_attrs)
-            mocked_subprocess.Popen.return_value = popen_mock
-            mocked_subprocess.PIPE = -1
-
-            call_rc = test_data["call_rc"]
-            mocked_subprocess.call.return_value = call_rc
-
-            caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
-            caclmgrd_daemon.update_control_plane_nat_acls('', {})
-            mocked_subprocess.Popen.assert_has_calls(test_data["expected_subprocess_calls"], any_order=True)
+        with mock.patch("caclmgrd.ControlPlaneAclManager.run_commands_pipe", return_value='sonic'):
+            with mock.patch("caclmgrd.subprocess") as mocked_subprocess:
+                popen_mock = mock.Mock()
+                popen_attrs = test_data["popen_attributes"]
+                popen_mock.configure_mock(**popen_attrs)
+                mocked_subprocess.Popen.return_value = popen_mock
+                mocked_subprocess.PIPE = -1
+
+                call_rc = test_data["call_rc"]
+                mocked_subprocess.call.return_value = call_rc
+
+                caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
+                caclmgrd_daemon.update_control_plane_nat_acls('', {})
+                mocked_subprocess.Popen.assert_has_calls(test_data["expected_subprocess_calls"], any_order=True)
 
     def test_get_ip_from_interface_table(self):
         if not os.path.exists(DBCONFIG_PATH):

tests/caclmgrd/caclmgrd_feature_test.py

@@ -0,0 +1,33 @@
+import os
+import sys
+import swsscommon
+from unittest.mock import call
+from unittest import TestCase, mock
+from tests.common.mock_configdb import MockConfigDb
+from sonic_py_common.general import load_module_from_source
+
+class TestCaclmgrd(TestCase):
+    def setUp(self):
+        swsscommon.swsscommon.ConfigDBConnector = MockConfigDb
+        test_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+        modules_path = os.path.dirname(test_path)
+        scripts_path = os.path.join(modules_path, "scripts")
+        sys.path.insert(0, modules_path)
+        caclmgrd_path = os.path.join(scripts_path, 'caclmgrd')
+        self.caclmgrd = load_module_from_source('caclmgrd', caclmgrd_path)
+
+    def test_run_commands_pipe(self):
+        caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
+        output = caclmgrd_daemon.run_commands_pipe(['echo', 'caclmgrd'], ['awk', '{print $1}'])
+        assert output == 'caclmgrd'
+
+        output = caclmgrd_daemon.run_commands_pipe([sys.executable, "-c", "import sys; sys.exit(6)"], [sys.executable, "-c", "import sys; sys.exit(8)"])
+        assert output == ''
+
+    def test_get_chain_list(self):
+        expected_calls = [call(['iptables', '-L', '-v', '-n'], ['grep', 'Chain'], ['awk', '{print $2}'])]
+        caclmgrd_daemon = self.caclmgrd.ControlPlaneAclManager("caclmgrd")
+        with mock.patch("caclmgrd.ControlPlaneAclManager.run_commands_pipe") as mock_run_commands_pipe:
+            caclmgrd_daemon.get_chain_list([], [''])
+            mock_run_commands_pipe.assert_has_calls(expected_calls)
+

tests/caclmgrd/caclmgrd_soc_rules_test.py

@@ -23,8 +23,8 @@
                 },
             },
             "expected_subprocess_calls": [
-                call("iptables -I INPUT 2 -p udp -m multiport --dports 3784,4784 -j ACCEPT", shell=True, universal_newlines=True, stdout=subprocess.PIPE),
-                call("ip6tables -I INPUT 2 -p udp -m multiport --dports 3784,4784 -j ACCEPT", shell=True, universal_newlines=True, stdout=subprocess.PIPE)
+                call(['iptables', '-I', 'INPUT', '2', '-p', 'udp', '-m', 'multiport', '--dports', '3784,4784', '-j', 'ACCEPT'], universal_newlines=True, stdout=subprocess.PIPE),
+                call(['ip6tables', '-I', 'INPUT', '2', '-p', 'udp', '-m', 'multiport', '--dports', '3784,4784', '-j', 'ACCEPT'], universal_newlines=True, stdout=subprocess.PIPE)
             ],
             "popen_attributes": {
                 'communicate.return_value': ('output', 'error'),