replace shell=True, replace xml, and replace exit() (sonic-net#2664)

Signed-off-by: maipbui <[email protected]>
#### What I did
The [xml.etree.ElementTree](https://docs.python.org/3/library/xml.etree.elementtree.html#module-xml.etree.ElementTree) module is not secure against maliciously constructed data.
`subprocess()` - when using with `shell=True` is dangerous. Using subprocess function without a static string can lead to command injection.
`sys.exit` is better than `exit`, considered good to use in production code.
Ref:
https://stackoverflow.com/questions/6501121/difference-between-exit-and-sys-exit-in-python
https://stackoverflow.com/questions/19747371/python-exit-commands-why-so-many-and-when-should-each-be-used
#### How I did it
Remove xml. Use [lxml](https://pypi.org/project/lxml/) XML parsers package that prevent potentially malicious operation.
`subprocess()` - use `shell=False` instead, use list of strings Ref: [https://semgrep.dev/docs/cheat-sheets/python-command-injection/#mitigation](https://semgrep.dev/docs/cheat-sheets/python-command-injection/#mitigation)
Replace `exit()` by `sys.exit()`
#### How to verify it
Pass UT
Manual test

Author: maipbui

pfcwd/main.py

@@ -243,7 +243,7 @@ def start(self, action, restoration_time, ports, detection_time):
             click.echo("Failed to run command, invalid options:")
             for opt in invalid_ports:
                 click.echo(opt)
-            exit(1)
+            sys.exit(1)
         self.start_cmd(action, restoration_time, ports, detection_time)
 
 
@@ -263,7 +263,7 @@ def verify_pfc_enable_status_per_port(self, port, pfcwd_info):
     @multi_asic_util.run_on_multi_asic
     def start_cmd(self, action, restoration_time, ports, detection_time):
         if os.geteuid() != 0:
-            exit("Root privileges are required for this operation")
+            sys.exit("Root privileges are required for this operation")
 
         all_ports = get_all_ports(
             self.db, self.multi_asic.current_namespace,
@@ -299,7 +299,7 @@ def start_cmd(self, action, restoration_time, ports, detection_time):
     @multi_asic_util.run_on_multi_asic
     def interval(self, poll_interval):
         if os.geteuid() != 0:
-            exit("Root privileges are required for this operation")
+            sys.exit("Root privileges are required for this operation")
         pfcwd_info = {}
         if poll_interval is not None:
             pfcwd_table = self.config_db.get_table(CONFIG_DB_PFC_WD_TABLE_NAME)
@@ -331,7 +331,7 @@ def interval(self, poll_interval):
                         poll_interval, entry_min_str
                     ), err=True
                 )
-                exit(1)
+                sys.exit(1)
 
             pfcwd_info['POLL_INTERVAL'] = poll_interval
             self.config_db.mod_entry(
@@ -341,7 +341,7 @@ def interval(self, poll_interval):
     @multi_asic_util.run_on_multi_asic
     def stop(self, ports):
         if os.geteuid() != 0:
-            exit("Root privileges are required for this operation")
+            sys.exit("Root privileges are required for this operation")
 
         all_ports = get_all_ports(
             self.db, self.multi_asic.current_namespace,
@@ -359,7 +359,7 @@ def stop(self, ports):
     @multi_asic_util.run_on_multi_asic
     def start_default(self):
         if os.geteuid() != 0:
-            exit("Root privileges are required for this operation")
+            sys.exit("Root privileges are required for this operation")
         enable = self.config_db.get_entry('DEVICE_METADATA', 'localhost').get(
             'default_pfcwd_status'
         )
@@ -394,15 +394,15 @@ def start_default(self):
     @multi_asic_util.run_on_multi_asic
     def counter_poll(self, counter_poll):
         if os.geteuid() != 0:
-            exit("Root privileges are required for this operation")
+            sys.exit("Root privileges are required for this operation")
         pfcwd_info = {}
         pfcwd_info['FLEX_COUNTER_STATUS'] = counter_poll
         self.config_db.mod_entry("FLEX_COUNTER_TABLE", "PFCWD", pfcwd_info)
 
     @multi_asic_util.run_on_multi_asic
     def big_red_switch(self, big_red_switch):
         if os.geteuid() != 0:
-            exit("Root privileges are required for this operation")
+            sys.exit("Root privileges are required for this operation")
         pfcwd_info = {}
         if big_red_switch is not None:
             pfcwd_info['BIG_RED_SWITCH'] = big_red_switch

scripts/boot_part

@@ -4,10 +4,10 @@
 import re
 import os
 import sys
-import commands
 import argparse
 import logging
 import tempfile
+from sonic_py_common.general import getstatusoutput_noshell
 
 logging.basicConfig(level=logging.WARN)
 logger = logging.getLogger(__name__)
@@ -20,7 +20,7 @@ re_hex = r'[0-9A-F]'
 ##   String - the standard output of the command, may be empty string
 def runcmd(cmd):
     logger.info('runcmd: {0}'.format(cmd))
-    rc, out = commands.getstatusoutput(cmd)
+    rc, out = getstatusoutput_noshell(cmd)
     if rc == 0:
         return out
     else:
@@ -29,7 +29,7 @@ def runcmd(cmd):
 
 def print_partitions(blkdev):
     assert blkdev
-    out = runcmd('sudo lsblk -r -o PARTLABEL,NAME')
+    out = runcmd(['sudo', 'lsblk', '-r', '-o', 'PARTLABEL,NAME'])
     ## Parse command output and print
     found_table = False
     for line in out.splitlines():
@@ -56,7 +56,7 @@ def print_partitions(blkdev):
 
 ## Get the current boot partition index
 def get_boot_partition(blkdev):
-    out = runcmd('cat /proc/mounts')
+    out = runcmd(['cat', '/proc/mounts'])
     if out is None: return None
 
     ## Parse command output and return the current boot partition index
@@ -76,16 +76,26 @@ def set_boot_partition(blkdev, index):
     devnode = blkdev + str(index)
     mntpath = tempfile.mkdtemp()
     try:
-        out = runcmd('sudo mount {0} {1}'.format(devnode, mntpath))
+        out = runcmd(['sudo', 'mount', devnode, mntpath])
         logger.info('mount out={0}'.format(out))
         if out is None: return
         ## Set GRUB bootable
-        out = runcmd('sudo grub-install --boot-directory="{0}" --recheck "{1}"'.format(mntpath, blkdev))
+        out = runcmd(['sudo', 'grub-install', '--boot-directory='+mntpath, "--recheck", blkdev])
         return out is not None
     finally:
         ## Cleanup
-        out = runcmd('sudo fuser -km {0} || sudo umount {0}'.format(mntpath))
-        logger.info('fuser out={0}'.format(out))
+        cmd1 = ['sudo', 'fuser', '-km', mntpath]
+        rc1, out1 = getstatusoutput_noshell(cmd1)
+        if rc1 != 0:
+            logger.error('Failed to run: {0}\n{1}'.format(cmd1, out1))
+            cmd2 = ['sudo', 'unmount', mntpath]
+            rc2, out2 = getstatusoutput_noshell(cmd2)
+            if rc2 == 0:
+                logger.info('Running command: {0}\n{1}'.format(' '.join(cmd2), out2))
+            else:
+                logger.error('Failed to run: {0}\n{1}'.format(cmd2, out2))
+        else:
+            logger.info('Running command: {0}\n{1}'.format(' '.join(cmd1), out1))
         os.rmdir(mntpath)
 
 def main():
@@ -100,7 +110,7 @@ def main():
         logger.setLevel(logging.INFO)
 
     ## Find ONIE partition and get the block device containing ONIE
-    out = runcmd("sudo blkid")
+    out = runcmd(["sudo", "blkid"])
     if not out: return -1
     for line in out.splitlines():
         m = re.match(r'/dev/(\w+)\d+: LABEL="ONIE-BOOT"', line)

scripts/check_db_integrity.py

@@ -36,9 +36,9 @@ def main():
 
     for db_name, schema in DB_SCHEMA.items():
         db_dump_file = "/tmp/{}.json".format(db_name)
-        dump_db_cmd = "sonic-db-dump -n 'COUNTERS_DB' -y > {}".format(db_dump_file)
-        p = subprocess.Popen(dump_db_cmd, shell=True, text=True,
-            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        dump_db_cmd = ["sonic-db-dump", "-n", 'COUNTERS_DB', "-y"]
+        with open(db_dump_file, 'w') as f:
+            p = subprocess.Popen(dump_db_cmd, text=True, stdout=f, stderr=subprocess.PIPE)
         (_, err) = p.communicate()
         rc = p.wait()
         if rc != 0:

scripts/configlet

@@ -1,4 +1,7 @@
 #!/usr/bin/env python3
+
+import sys
+
 """ JSON based configlet update
 
 A tool to update CONFIG-DB with JSON diffs that can update/delete redis-DB.
@@ -195,7 +198,7 @@ def main():
     if not do_act:
         print("Expect an action update/delete or for debug parse/test\n")
         parser.print_help()
-        exit(-1)
+        sys.exit(-1)
 
     for json_file in args.json:
         with open(json_file, 'r') as stream:

scripts/disk_check.py

@@ -83,7 +83,7 @@ def test_writable(dirs):
 
 
 def run_cmd(cmd):
-    proc = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE)
+    proc = subprocess.run(cmd, shell=False, stdout=subprocess.PIPE)
     ret = proc.returncode
     if ret:
         log_err("failed: ret={} cmd={}".format(ret, cmd))
@@ -120,9 +120,8 @@ def do_mnt(dirs):
         os.mkdir(d_upper)
         os.mkdir(d_work)
 
-        ret = run_cmd("mount -t overlay overlay_{} -o lowerdir={},"
-        "upperdir={},workdir={} {}".format(
-            d_name, d, d_upper, d_work, d))
+        ret = run_cmd(["mount", "-t", "overlay", "overlay_{}".format(d_name),\
+                "-o", "lowerdir={},upperdir={},workdir={}".format(d, d_upper, d_work), d])
         if ret:
             break
 

scripts/dropconfig

@@ -375,25 +375,25 @@ Examples:
                                    reasons)
         except InvalidArgumentError as err:
             print('Encountered error trying to install counter: {}'.format(err.message))
-            exit(1)
+            sys.exit(1)
     elif command == 'uninstall':
         try:
             dconfig.delete_counter(name)
         except InvalidArgumentError as err:
             print('Encountered error trying to uninstall counter: {}'.format(err.message))
-            exit(1)
+            sys.exit(1)
     elif command == 'add':
         try:
             dconfig.add_reasons(name, reasons)
         except InvalidArgumentError as err:
             print('Encountered error trying to add reasons: {}'.format(err.message))
-            exit(1)
+            sys.exit(1)
     elif command == 'remove':
         try:
             dconfig.remove_reasons(name, reasons)
         except InvalidArgumentError as err:
             print('Encountered error trying to remove reasons: {}'.format(err.message))
-            exit(1)
+            sys.exit(1)
     elif command == 'show_config':
         dconfig.print_counter_config(group)
     elif command == 'show_capabilities':

scripts/dump_nat_entries.py

@@ -8,8 +8,10 @@
 import subprocess
 
 def main():
-    ctdumpcmd = 'conntrack -L -j > /host/warmboot/nat/nat_entries.dump'
-    p = subprocess.Popen(ctdumpcmd, shell=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    ctdumpcmd = ['conntrack', '-L', '-j'] 
+    file = '/host/warmboot/nat/nat_entries.dump'
+    with open(file, 'w') as f:
+        p = subprocess.Popen(ctdumpcmd, text=True, stdout=f, stderr=subprocess.PIPE)
     (output, err) = p.communicate()
     rc = p.wait()
 

scripts/ipintutil

@@ -72,13 +72,12 @@ def get_if_admin_state(iface, namespace):
     """
     Given an interface name, return its admin state reported by the kernel
     """
-    cmd = "cat /sys/class/net/{0}/flags".format(iface)
+    cmd = ["cat", "/sys/class/net/{0}/flags".format(iface)]
     if namespace != constants.DEFAULT_NAMESPACE:
-        cmd = "sudo ip netns exec {} {}".format(namespace, cmd)
+        cmd = ["sudo", "ip", "netns", "exec", namespace] + cmd
     try:
         proc = subprocess.Popen(
             cmd,
-            shell=True,
             stderr=subprocess.STDOUT,
             stdout=subprocess.PIPE,
             text=True)
@@ -105,13 +104,12 @@ def get_if_oper_state(iface, namespace):
     """
     Given an interface name, return its oper state reported by the kernel.
     """
-    cmd = "cat /sys/class/net/{0}/carrier".format(iface)
+    cmd = ["cat", "/sys/class/net/{0}/carrier".format(iface)]
     if namespace != constants.DEFAULT_NAMESPACE:
-        cmd = "sudo ip netns exec {} {}".format(namespace, cmd)
+        cmd = ["sudo", "ip", "netns", "exec", namespace] + cmd
     try:
         proc = subprocess.Popen(
             cmd,
-            shell=True,
             stderr=subprocess.STDOUT,
             stdout=subprocess.PIPE,
             text=True)

scripts/lldpshow

@@ -23,7 +23,7 @@ import argparse
 import re
 import subprocess
 import sys
-import xml.etree.ElementTree as ET
+from lxml import etree as ET
 
 from sonic_py_common import device_info
 from swsscommon.swsscommon import ConfigDBConnector
@@ -80,9 +80,14 @@ class Lldpshow(object):
             lldp_interface_list = lldp_port if lldp_port is not None else self.lldp_interface[lldp_instace_num]
             # In detail mode we will pass interface list (only front ports) and get O/P as plain text
             # and in table format we will get xml output
-            lldp_cmd = 'sudo docker exec -i lldp{} lldpctl '.format(self.lldp_instance[lldp_instace_num]) + (
-                '-f xml' if not lldp_detail_info else lldp_interface_list)
-            p = subprocess.Popen(lldp_cmd, stdout=subprocess.PIPE, shell=True, text=True)
+            if not lldp_detail_info:
+                lldp_args = ['-f', 'xml']
+            elif lldp_interface_list == '':
+                lldp_args = []
+            else:
+                lldp_args = [lldp_interface_list]
+            lldp_cmd = ['sudo', 'docker', 'exec', '-i', 'lldp{}'.format(self.lldp_instance[lldp_instace_num]), 'lldpctl'] + lldp_args
+            p = subprocess.Popen(lldp_cmd, stdout=subprocess.PIPE, text=True)
             (output, err) = p.communicate()
             ## Wait for end of command. Get return returncode ##
             returncode = p.wait()
@@ -121,7 +126,7 @@ class Lldpshow(object):
         if lldp_detail_info:
             return
         for lldpraw in self.lldpraw:
-            neis = ET.fromstring(lldpraw)
+            neis = ET.fromstring(lldpraw.encode())
             intfs = neis.findall('interface')
             for intf in intfs:
                 l_intf = intf.attrib['name']

scripts/storyteller

@@ -11,6 +11,7 @@ import subprocess
 import sys
 
 from shlex import quote
+from sonic_py_common.general import getstatusoutput_noshell_pipe
 
 regex_dict = {
                 'acl'       : r'acl\|ACL\|Acl',
@@ -28,31 +29,35 @@ reference_file = '/tmp/storyteller_time_reference'
 
 def exec_cmd(cmd):
     # Use universal_newlines (instead of text) so that this tool can work with any python versions.
-    out = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=True, universal_newlines=True)
+    out = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True)
     stdout, stderr = out.communicate()
     return out.returncode, stdout, stderr
 
 
 def build_options(after=0, before=0, context=0):
     options = []
     if after:
-        options.append('-A {}'.format(after))
+        options += ['-A', str(after)]
     if before:
-        options.append('-B {}'.format(before))
+        options += ['-B', str(before)]
     if context:
-        options.append('-C {}'.format(context))
+        options += ['-C', str(context)]
 
-    return ' '.join(x for x in options)
+    return options
 
 
 def find_log(logpath, log, regex, after=0, before=0, context=0, field=0):
     options = build_options(after, before, context)
     if field <= 0:
-        cmd = 'find -L {}/{}* -newer {} | xargs ls -rt | xargs zgrep -a {} "{}"'.format(logpath, log, reference_file, options, regex)
+        cmd0 = ['find', logpath, "-name", "{}*".format(log), "-newer", reference_file]
+        cmd1 = ["xargs", "ls", "-rt"]
+        cmd2 = ["xargs", "zgrep", "-a"] + options + [regex]
     else:
-        cmd = 'find -L {0}/{1}* -newer {2} | sort -rn -t . -k {3},{3} | xargs zgrep -a {4} "{5}"'.format(logpath, log, reference_file, field, options, regex)
+        cmd0 = ['find', logpath, "-name", "{}*".format(log), "-newer", reference_file]
+        cmd1 = ["sort", "-rn", "-t", ".", "-k", "{0},{0}".format(field)]
+        cmd2 = ["xargs", "zgrep", "-a"] + options + [regex]
 
-    _, out, _ = exec_cmd(cmd)
+    _, out = getstatusoutput_noshell_pipe(cmd0, cmd1, cmd2)
     '''
         Opportunity to improve:
             output (out) can be split to lines and send to a filter to
@@ -71,12 +76,12 @@ def build_regex(category):
 
 
 def configure_time_filter(since):
-    ret_code, _, _ = exec_cmd('date --date {}'.format(since))
+    ret_code, _, _ = exec_cmd(['date', '--date', since])
     if ret_code:
         print('invalid date "{}"'.format(since))
         sys.exit(1)
 
-    exec_cmd('touch --date "{}" {}'.format(since, reference_file))
+    exec_cmd(['touch', '--date', since, reference_file])
 
 
 def main():