chroot / means no chroot

Author: penguinolog

exec_helpers/_ssh_client_base.py

@@ -1323,11 +1323,11 @@ def get_result(remote: "SSHClientBase") -> exec_result.ExecResult:
             # pylint: disable=protected-access
             cmd_for_log: str = remote._mask_command(cmd=command, log_mask_re=log_mask_re)
 
+            target_path: typing.Optional[str] = kwargs.get("chroot_path", remote._chroot_path)
+            chroot_info: str = "" if not target_path or target_path == "/" else f" (with chroot to: {target_path!r})"
+
             remote.logger.log(
-                level=log_level,
-                msg=f"Executing command"
-                f"{'' if not remote._chroot_path else f' (with chroot to: {remote._chroot_path})'}:\n"
-                f"{cmd_for_log!r}\n",
+                level=log_level, msg=f"Executing command{chroot_info}:\n{cmd_for_log!r}\n",
             )
             async_result: SshExecuteAsyncResult = remote._execute_async(
                 command,

exec_helpers/api.py

@@ -236,10 +236,11 @@ def _prepare_command(self, cmd: str, chroot_path: typing.Optional[str] = None) -
         :param chroot_path: path to make chroot for execution
         :return: final command, includes chroot, if required
         """
-        if any((chroot_path, self._chroot_path)):
-            target_path: str = shlex.quote(chroot_path if chroot_path else self._chroot_path)  # type: ignore
+        target_path: typing.Optional[str] = chroot_path if chroot_path else self._chroot_path
+        if target_path and target_path != "/":
+            chroot_dst: str = shlex.quote(target_path.strip())
             quoted_command = shlex.quote(cmd)
-            return f'chroot {target_path} sh -c {shlex.quote(f"eval {quoted_command}")}'
+            return f'chroot {chroot_dst} sh -c {shlex.quote(f"eval {quoted_command}")}'
         return cmd
 
     @abc.abstractmethod
@@ -363,10 +364,11 @@ def execute(
         cmd_for_log: str = self._mask_command(cmd=command, log_mask_re=log_mask_re)
         log_level: int = logging.INFO if verbose else logging.DEBUG
 
+        target_path: typing.Optional[str] = kwargs.get("chroot_path", self._chroot_path)
+        chroot_info: str = "" if not target_path or target_path == "/" else f" (with chroot to: {target_path!r})"
+
         self.logger.log(
-            level=log_level,
-            msg=f"Executing command{'' if not self._chroot_path else f' (with chroot to: {self._chroot_path})'}:\n"
-            f"{cmd_for_log!r}\n",
+            level=log_level, msg=f"Executing command{chroot_info}:\n{cmd_for_log!r}\n",
         )
 
         async_result: ExecuteAsyncResult = self._execute_async(

exec_helpers/async_api/api.py

@@ -220,10 +220,11 @@ async def execute(  # type: ignore
         cmd_for_log: str = self._mask_command(cmd=command, log_mask_re=log_mask_re)
         log_level: int = logging.INFO if verbose else logging.DEBUG
 
+        target_path: typing.Optional[str] = kwargs.get("chroot_path", self._chroot_path)
+        chroot_info: str = "" if not target_path or target_path == "/" else f" (with chroot to: {target_path!r})"
+
         self.logger.log(
-            level=log_level,
-            msg=f"Executing command{'' if not self._chroot_path else f' (with chroot to: {self._chroot_path})'}:\n"
-            f"{cmd_for_log!r}\n",
+            level=log_level, msg=f"Executing command{chroot_info}:\n{cmd_for_log!r}\n",
         )
 
         async_result: api.ExecuteAsyncResult = await self._execute_async(

test/test_ssh_client_execute_async_special.py

@@ -218,30 +218,30 @@ def test_010_check_stdin_closed(paramiko_ssh_client, chan_makefile, auto_add_pol
 
 def test_011_execute_async_chroot_cmd(ssh, ssh_transport_channel):
     """Command-only chroot path."""
-    ssh._execute_async(command, chroot_path="/")
+    ssh._execute_async(command, chroot_path="/mnt")
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),
-            mock.call.exec_command(f'chroot / sh -c {shlex.quote(f"eval {quoted_command}")}\n'),
+            mock.call.exec_command(f'chroot /mnt sh -c {shlex.quote(f"eval {quoted_command}")}\n'),
         )
     )
 
 
 def test_012_execute_async_chroot_context(ssh, ssh_transport_channel):
     """Context-managed chroot path."""
-    with ssh.chroot("/"):
+    with ssh.chroot("/mnt"):
         ssh._execute_async(command)
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),
-            mock.call.exec_command(f'chroot / sh -c {shlex.quote(f"eval {quoted_command}")}\n'),
+            mock.call.exec_command(f'chroot /mnt sh -c {shlex.quote(f"eval {quoted_command}")}\n'),
         )
     )
 
 
 def test_013_execute_async_no_chroot_context(ssh, ssh_transport_channel):
     """Context-managed chroot path override."""
-    ssh._chroot_path = "/"
+    ssh._chroot_path = "/mnt"
 
     with ssh.chroot(None):
         ssh._execute_async(command)
@@ -250,12 +250,12 @@ def test_013_execute_async_no_chroot_context(ssh, ssh_transport_channel):
 
 def test_012_execute_async_chroot_path(ssh, ssh_transport_channel):
     """Command-only chroot path."""
-    with ssh.chroot(pathlib.Path("/")):
+    with ssh.chroot(pathlib.Path("/mnt")):
         ssh._execute_async(command)
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),
-            mock.call.exec_command(f'chroot / sh -c {shlex.quote(f"eval {quoted_command}")}\n'),
+            mock.call.exec_command(f'chroot /mnt sh -c {shlex.quote(f"eval {quoted_command}")}\n'),
         )
     )
 
@@ -266,3 +266,12 @@ def test_013_execute_async_chroot_path_invalid_type(ssh, ssh_transport_channel):
         with ssh.chroot(...):
             ssh._execute_async(command)
         assert str(exc.value) == f"path={...!r} is not instance of Optional[Union[str, pathlib.Path]]"
+
+
+def test_014_execute_async_no_chroot_root(ssh, ssh_transport_channel):
+    """Context-managed chroot path override."""
+    ssh._chroot_path = "/mnt"
+
+    with ssh.chroot("/"):
+        ssh._execute_async(command)
+    ssh_transport_channel.assert_has_calls((mock.call.makefile_stderr("rb"), mock.call.exec_command(f"{command}\n")))