[sonic-py-common] Add getstatusoutput_noshell() functions to general module

src/sonic-py-common/sonic_py_common/general.py

@@ -1,4 +1,5 @@
 import sys
+from subprocess import Popen, STDOUT, PIPE, CalledProcessError, check_output
 
 
 def load_module_from_source(module_name, file_path):
@@ -23,3 +24,56 @@ def load_module_from_source(module_name, file_path):
     sys.modules[module_name] = module
 
     return module
+
+
+def getstatusoutput_noshell(cmd):
+    """
+    This function implements getstatusoutput API from subprocess module
+    but using shell=False to prevent shell injection.
+    Ref: https://github.com/python/cpython/blob/3.10/Lib/subprocess.py#L602
+    """
+    try:
+        output = check_output(cmd, text=True, stderr=STDOUT)
+        exitcode = 0
+    except CalledProcessError as ex:
+        output = ex.output
+        exitcode = ex.returncode
+    if output[-1:] == '\n':
+        output = output[:-1]
+    return exitcode, output
+
+
+def getstatusoutput_noshell_pipe(cmd1, cmd2):
+    """
+    This function implements getstatusoutput API from subprocess module
+    but using shell=False to prevent shell injection. Input command includes
+    pipe command.
+    """
+    with Popen(cmd1, universal_newlines=True, stdout=PIPE) as p1:
+        with Popen(cmd2, universal_newlines=True, stdin=p1.stdout, stdout=PIPE) as p2:
+            output = p2.communicate()[0]
+            p1.wait()
+            if p1.returncode != 0 or p2.returncode != 0:
+                return ([p1.returncode, p2.returncode], output)
+    if output[-1:] == '\n':
+        output = output[:-1]
+    return ([0, 0], output)
+
+
+def check_output_pipe(cmd1, cmd2):
+    """
+    This function implements check_output API from subprocess module.
+    Input command includes pipe command.
+    """
+    cmd = ' '.join(cmd1) + ' | ' + ' '.join(cmd2)
+    with Popen(cmd1, universal_newlines=True, stdout=PIPE) as p1:
+        with Popen(cmd2, universal_newlines=True, stdin=p1.stdout, stdout=PIPE) as p2:
+            output = p2.communicate()[0]
+            p1.wait()
+            if p1.returncode != 0 or p2.returncode != 0:
+                if p1.returncode != 0:
+                    raise CalledProcessError(returncode=p1.returncode, cmd=cmd, output=output)
+                else:
+                    raise CalledProcessError(returncode=p2.returncode, cmd=cmd, output=output)
+    return output
+

src/sonic-py-common/tests/test_general.py

@@ -0,0 +1,33 @@
+import os
+import sys
+import pytest
+import subprocess
+from sonic_py_common.general import getstatusoutput_noshell, getstatusoutput_noshell_pipe, check_output_pipe
+
+
+def test_getstatusoutput_noshell(tmpdir):
+    exitcode, output = getstatusoutput_noshell(['echo', 'sonic'])
+    assert (exitcode, output) == (0, 'sonic')
+
+    name = os.path.join(tmpdir, 'foo')
+    exitcode, output = getstatusoutput_noshell(['cat', name])
+    assert exitcode != 0
+
+def test_getstatusoutput_noshell_pipe():
+    exitcode, output = getstatusoutput_noshell_pipe(['echo', 'sonic'], ['awk', '{print $1}'])
+    assert (exitcode, output) == ([0, 0], 'sonic')
+
+    exitcode, output = getstatusoutput_noshell_pipe([sys.executable, "-c", "import sys; sys.exit(6)"], [sys.executable, "-c", "import sys; sys.exit(8)"])
+    assert exitcode == [6, 8]
+
+def test_check_output_pipe():
+    output = check_output_pipe(['echo', 'sonic'], ['awk', '{print $1}'])
+    assert output == 'sonic\n'
+
+    with pytest.raises(subprocess.CalledProcessError) as e:
+        check_output_pipe([sys.executable, "-c", "import sys; sys.exit(6)"], [sys.executable, "-c", "import sys; sys.exit(0)"])
+        assert e.returncode == [6, 0]
+
+    with pytest.raises(subprocess.CalledProcessError) as e:
+        check_output_pipe([sys.executable, "-c", "import sys; sys.exit(0)"], [sys.executable, "-c", "import sys; sys.exit(6)"])
+        assert e.returncode == [0, 6]