diff options
| -rw-r--r-- | python/vyos/utils/process.py | 44 | ||||
| -rw-r--r-- | src/tests/test_utils.py | 31 |
2 files changed, 75 insertions, 0 deletions
diff --git a/python/vyos/utils/process.py b/python/vyos/utils/process.py index 28027684c..3e5576925 100644 --- a/python/vyos/utils/process.py +++ b/python/vyos/utils/process.py @@ -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}} |