Use shlex.quote to prevent interrupting chroot path and shell expand …

…out of chroot

Signed-off-by: Aleksei Stepanov <penguinolog@gmail.com>

exec_helpers/_ssh_client_base.py

@@ -399,6 +399,20 @@ def keepalive(self, enforce: bool = True) -> "typing.ContextManager[None]":
         """
         return _KeepAliveContext(ssh=self, enforce=enforce)
 
+    def _prepare_command(self, cmd: str, chroot_path: typing.Optional[str] = None) -> str:
+        """Prepare command: cower chroot and other cases.
+
+        :param cmd: main command
+        :param chroot_path: path to make chroot for execution
+        :returns: final command, includes chroot, if required
+        """
+        if not self.sudo_mode:
+            return super()._prepare_command(cmd=cmd, chroot_path=chroot_path)
+        if any((chroot_path, self._chroot_path)):
+            target_path: str = shlex.quote(chroot_path if chroot_path else self._chroot_path)
+            return f'chroot {target_path} sudo sh -c "eval {shlex.quote(cmd)}"'
+        return f'sudo -S sh -c "eval {shlex.quote(cmd)}"'
+
     # noinspection PyMethodOverriding
     def _execute_async(  # pylint: disable=arguments-differ
         self,
@@ -479,7 +493,6 @@ def _execute_async(  # pylint: disable=arguments-differ
 
         started = datetime.datetime.utcnow()
         if self.sudo_mode:
-            cmd = f'sudo -S bash -c "eval {shlex.quote(cmd)}"'
             chan.exec_command(cmd)  # nosec  # Sanitize on caller side
             if stdout.channel.closed is False:
                 # noinspection PyTypeChecker

exec_helpers/api.py

@@ -26,6 +26,7 @@
 import datetime
 import logging
 import re
+import shlex
 import threading
 import typing
 import warnings
@@ -227,7 +228,8 @@ def _prepare_command(self, cmd: str, chroot_path: typing.Optional[str] = None) -
         :returns: final command, includes chroot, if required
         """
         if any((chroot_path, self._chroot_path)):
-            return f"chroot {chroot_path if chroot_path else self._chroot_path} {cmd}"
+            target_path: str = shlex.quote(chroot_path if chroot_path else self._chroot_path)
+            return f'chroot {target_path} sh -c "eval {shlex.quote(cmd)}"'
         return cmd
 
     def execute_async(  # pylint: disable=missing-param-doc,differing-param-doc,differing-type-doc

test/test_ssh_client_execute_async_special.py

@@ -120,7 +120,7 @@ def test_001_execute_async_sudo(ssh, ssh_transport_channel):
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),
-            mock.call.exec_command(f'sudo -S bash -c \"eval {shlex.quote(cmd_execute)}\"'),
+            mock.call.exec_command(f'sudo -S sh -c \"eval {shlex.quote(command)}\"\n'),
         )
     )
 
@@ -133,7 +133,7 @@ def test_002_execute_async_with_sudo_enforce(ssh, ssh_transport_channel):
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),
-            mock.call.exec_command(f'sudo -S bash -c \"eval {shlex.quote(cmd_execute)}\"'),
+            mock.call.exec_command(f'sudo -S sh -c \"eval {shlex.quote(command)}\"\n'),
         )
     )
 
@@ -163,7 +163,7 @@ def test_005_execute_async_sudo_password(ssh, ssh_transport_channel, mocker):
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),
-            mock.call.exec_command(f'sudo -S bash -c \"eval {shlex.quote(cmd_execute)}\"'),
+            mock.call.exec_command(f'sudo -S sh -c \"eval {shlex.quote(command)}\"\n'),
         )
     )
 
@@ -233,7 +233,7 @@ def test_011_execute_async_chroot_cmd(ssh, ssh_transport_channel):
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),
-            mock.call.exec_command(f'chroot / {command}\n'),
+            mock.call.exec_command(f'chroot / sh -c "eval {shlex.quote(command)}"\n'),
         )
     )
 
@@ -245,7 +245,7 @@ def test_012_execute_async_chroot_context(ssh, ssh_transport_channel):
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),
-            mock.call.exec_command(f'chroot / {command}\n'),
+            mock.call.exec_command(f'chroot / sh -c "eval {shlex.quote(command)}"\n'),
         )
     )