Merge pull request #5285 from c-po/cmdl

utils: T9003: add list-argument variant of cmd() for safer subprocess execution

2 files changed, 78 insertions, 3 deletions

diff --git a/python/vyos/utils/process.py b/python/vyos/utils/process.py
index e78c0b969..7b6f41a99 100644
--- a/python/vyos/utils/process.py
+++ b/python/vyos/utils/process.py
@@ -23,7 +23,7 @@ from subprocess import STDOUT
 from subprocess import DEVNULL
 
 def get_wrapper(vrf, netns):
-    wrapper = None
+    wrapper = []
     if vrf:
         wrapper = ['ip', 'vrf', 'exec', vrf]
     elif netns:
@@ -176,7 +176,7 @@ def run(command, flag='', shell=None, input=None, timeout=None, env=None,
 
 
 def cmd(command, flag='', shell=None, input=None, timeout=None, env=None,
-        stdout=PIPE, stderr=PIPE, decode='utf-8', raising=None, message='',
+        stdout=PIPE, stderr=PIPE, raising=None, message='',
         expect=[0], vrf=None, netns=None):
     """
     A wrapper around popen, which returns the stdout and
@@ -192,7 +192,7 @@ def cmd(command, flag='', shell=None, input=None, timeout=None, env=None,
         stdout=stdout, stderr=stderr,
         input=input, timeout=timeout,
         env=env, shell=shell,
-        decode=decode,
+        decode='utf-8',
         vrf=vrf,
         netns=netns,
     )
@@ -210,6 +210,50 @@ def cmd(command, flag='', shell=None, input=None, timeout=None, env=None,
             raise raising(feedback)
     return decoded
 
+def cmdl(command: list[str], flag: str = '', input: str | bytes | None = None,
+         timeout: float | None = None, env: dict[str, str] | None = None,
+         stdout: int = PIPE, stderr: int = PIPE,
+         raising: type[Exception] | None = None, message: str = '',
+         expect: list[int] | None = None, vrf: str | None = None,
+         netns: str | None = None, sudo: bool = False) -> str:
+    """
+    A list-argument variant of cmd() for safer subprocess execution.
+
+    command must be a list of strings; no shell interpolation is performed,
+    which eliminates a class of command-injection risks present when building
+    commands with f-strings or other string formatting.
+
+    sudo: prepend sudo(8) to the command for privileged execution
+    """
+    if not isinstance(command, list):
+        raise TypeError(f'cmdl() requires a list, got {type(command).__name__}')
+    if expect is None:
+        expect = [0]
+    if sudo:
+        command = ['sudo'] + command
+    decoded, code = popen(
+        command, flag,
+        shell=False,
+        input=input, timeout=timeout,
+        env=env,
+        stdout=stdout, stderr=stderr,
+        decode='utf-8',
+        vrf=vrf, netns=netns,
+    )
+    if code not in expect:
+        wrapper = get_wrapper(vrf, netns)
+        cmd_str = shlex.join(wrapper + command)
+        feedback = message + '\n' if message else ''
+        feedback += f'failed to run command: {cmd_str}\n'
+        feedback += f'returned: {decoded}\n'
+        feedback += f'exit code: {code}'
+        if raising is None:
+            raise OSError(code, feedback)
+        else:
+            raise raising(feedback)
+    return decoded
+
+
 def rc_cmd(command, flag='', shell=None, input=None, timeout=None, env=None,
            stdout=PIPE, stderr=STDOUT, decode='utf-8', vrf=None, netns=None,
            buffered=True):
diff --git a/src/tests/test_utils.py b/src/tests/test_utils.py
index 5022b24f6..432672f06 100644
--- a/src/tests/test_utils.py
+++ b/src/tests/test_utils.py
@@ -16,6 +16,37 @@ from unittest import TestCase
 from unittest.mock import patch
 
 class TestVyOSUtils(TestCase):
+    def test_cmdl_basic(self):
+        from vyos.utils.process import cmdl
+        out = cmdl(['echo', 'hello'])
+        self.assertEqual(out, 'hello')
+
+    def test_cmdl_requires_list(self):
+        from vyos.utils.process import cmdl
+        with self.assertRaises(TypeError):
+            cmdl('echo hello')
+
+    def test_cmdl_sudo_prepends(self):
+        from vyos.utils.process import cmdl
+        with patch('vyos.utils.process.popen') as mock_popen:
+            mock_popen.return_value = ('', 0)
+            cmdl(['id'], sudo=True)
+            args, kwargs = mock_popen.call_args
+            self.assertEqual(args[0][0], 'sudo')
+            self.assertEqual(args[0][1], 'id')
+
+    def test_cmdl_no_shell(self):
+        from vyos.utils.process import cmdl
+        with patch('vyos.utils.process.popen') as mock_popen:
+            mock_popen.return_value = ('', 0)
+            cmdl(['true'])
+            _, kwargs = mock_popen.call_args
+            self.assertEqual(kwargs.get('shell'), False)
+
+    def test_cmdl_raises_on_nonzero(self):
+        from vyos.utils.process import cmdl
+        with self.assertRaises(OSError):
+            cmdl(['false'])
     def test_key_mangling(self):
         from vyos.utils.dict import mangle_dict_keys
         data = {"foo-bar": {"baz-quux": None}}