Allow to configure chroot command (#155)

Allow the chroot command to be configured to something other than "chroot".

Co-authored-by: Geert Kloosterman <[email protected]>

Author: gjkloosterman

exec_helpers/_helpers.py

@@ -78,18 +78,20 @@ def cmd_to_string(command: str | Iterable[str]) -> str:
     return shlex.join(command)
 
 
-def chroot_command(command: str, chroot_path: str | None = None) -> str:
+def chroot_command(command: str, chroot_path: str | None = None, chroot_exe: str = "chroot") -> str:
     """Prepare command for chroot execution.
 
     :param command: original command.
     :type command: str
     :param chroot_path: chroot path
     :type chroot_path: str | None
+    :param chroot_exe: chroot executable
+    :type chroot_exe: str
     :return: command to be executed with chroot rules if applicable
     :rtype: str
     """
     if chroot_path and chroot_path != "/":
         chroot_dst: str = shlex.quote(chroot_path.strip())
         quoted_command = shlex.quote(command)
-        return f'chroot {chroot_dst} sh -c {shlex.quote(f"eval {quoted_command}")}'
+        return f'{chroot_exe} {chroot_dst} sh -c {shlex.quote(f"eval {quoted_command}")}'
     return command

exec_helpers/_ssh_base.py

@@ -981,23 +981,24 @@ def keepalive(self, enforce: KeepAlivePeriodT = 1) -> _KeepAliveContext:
         """
         return _KeepAliveContext(ssh=self, enforce=int(enforce))
 
-    def _prepare_command(self, cmd: str, chroot_path: str | None = None) -> str:
-        """Prepare command: cower chroot and other cases.
+    def _prepare_command(self, cmd: str, chroot_path: str | None = None, chroot_exe: str = "chroot") -> str:
+        """Prepare command: cover chroot and other cases.
 
         :param cmd: main command
         :param chroot_path: path to make chroot for execution
+        :param chroot_exe: chroot executable, default "chroot"
         :return: final command, includes chroot, if required
         """
         if not self.sudo_mode:
-            return super()._prepare_command(cmd=cmd, chroot_path=chroot_path)
+            return super()._prepare_command(cmd=cmd, chroot_path=chroot_path, chroot_exe=chroot_exe)
         quoted_command: str = shlex.quote(cmd)
         if chroot_path is self._chroot_path is None:
             return f'sudo -S sh -c {shlex.quote(f"eval {quoted_command}")}'
         if chroot_path is not None:
             target_path: str = shlex.quote(chroot_path)
         else:
             target_path = shlex.quote(self._chroot_path)  # type: ignore[arg-type]
-        return f'chroot {target_path} sudo sh -c {shlex.quote(f"eval {quoted_command}")}'
+        return f'{chroot_exe} {target_path} sudo sh -c {shlex.quote(f"eval {quoted_command}")}'
 
     def _exec_command(  # type: ignore[override]
         self,
@@ -1094,6 +1095,7 @@ def open_execute_context(
         open_stdout: bool = True,
         open_stderr: bool = True,
         chroot_path: str | None = None,
+        chroot_exe: str = "chroot",
         get_pty: bool = False,
         width: int = 80,
         height: int = 24,
@@ -1112,6 +1114,8 @@ def open_execute_context(
         :type open_stderr: bool
         :param chroot_path: chroot path override
         :type chroot_path: str | None
+        :param chroot_exe: chroot exe override
+        :type chroot_exe: str
         :param get_pty: Get PTY for connection
         :type get_pty: bool
         :param width: PTY width
@@ -1128,7 +1132,7 @@ def open_execute_context(
         """
         return _SSHExecuteContext(
             transport=self._ssh_transport,
-            command=f"{self._prepare_command(cmd=command, chroot_path=chroot_path)}\n",
+            command=f"{self._prepare_command(cmd=command, chroot_path=chroot_path, chroot_exe=chroot_exe)}\n",
             stdin=None if stdin is None else self._string_bytes_bytearray_as_bytes(stdin),
             open_stdout=open_stdout,
             open_stderr=open_stderr,
@@ -1155,6 +1159,7 @@ def execute(
         open_stderr: bool = True,
         log_stderr: bool = True,
         chroot_path: str | None = None,
+        chroot_exe: str = "chroot",
         get_pty: bool = False,
         width: int = 80,
         height: int = 24,
@@ -1183,6 +1188,8 @@ def execute(
         :type log_stderr: bool
         :param chroot_path: chroot path override
         :type chroot_path: str | None
+        :param chroot_exe: chroot exe override
+        :type chroot_exe: str
         :param get_pty: Get PTY for connection
         :type get_pty: bool
         :param width: PTY width
@@ -1199,6 +1206,7 @@ def execute(
         .. versionchanged:: 2.1.0 Allow parallel calls
         .. versionchanged:: 7.0.0 Allow command as list of arguments. Command will be joined with components escaping.
         .. versionchanged:: 8.0.0 chroot path exposed.
+        .. versionchanged:: 8.1.0 chroot exe added.
         """
         return super().execute(
             command=command,
@@ -1211,6 +1219,7 @@ def execute(
             open_stderr=open_stderr,
             log_stderr=log_stderr,
             chroot_path=chroot_path,
+            chroot_exe=chroot_exe,
             get_pty=get_pty,
             width=width,
             height=height,
@@ -1230,6 +1239,7 @@ def __call__(
         open_stderr: bool = True,
         log_stderr: bool = True,
         chroot_path: str | None = None,
+        chroot_exe: str = "chroot",
         get_pty: bool = False,
         width: int = 80,
         height: int = 24,
@@ -1258,6 +1268,8 @@ def __call__(
         :type log_stderr: bool
         :param chroot_path: chroot path override
         :type chroot_path: str | None
+        :param chroot_exe: chroot exe override
+        :type chroot_exe: str
         :param get_pty: Get PTY for connection
         :type get_pty: bool
         :param width: PTY width
@@ -1645,6 +1657,7 @@ def execute_together(
         open_stdout: bool = True,
         open_stderr: bool = True,
         chroot_path: str | None = None,
+        chroot_exe: str = "chroot",
         verbose: bool = False,
         log_mask_re: LogMaskReT = None,
         exception_class: type[exceptions.ParallelCallProcessError] = exceptions.ParallelCallProcessError,
@@ -1670,6 +1683,8 @@ def execute_together(
         :type open_stderr: bool
         :param chroot_path: chroot path override
         :type chroot_path: str | None
+        :param chroot_exe: chroot exe override
+        :type chroot_exe: str
         :param verbose: produce verbose log record on command call
         :type verbose: bool
         :param log_mask_re: regex lookup rule to mask command for logger.
@@ -1705,6 +1720,7 @@ def get_result(remote: SSHClientBase) -> exec_result.ExecResult:
                 log_mask_re=log_mask_re,
                 log_level=log_level,
                 chroot_path=chroot_path,
+                chroot_exe=chroot_exe,
                 **kwargs,
             )
             # pylint: enable=protected-access
@@ -1715,6 +1731,7 @@ def get_result(remote: SSHClientBase) -> exec_result.ExecResult:
                 open_stdout=open_stdout,
                 open_stderr=open_stderr,
                 chroot_path=chroot_path,
+                chroot_exe=chroot_exe,
                 timeout=timeout,
                 **kwargs,
             ) as async_result:

exec_helpers/api.py

@@ -172,31 +172,40 @@ class _ChRootContext(typing.ContextManager[None]):
     :type conn: ExecHelper
     :param path: chroot path or None for no chroot
     :type path: str | pathlib.Path | None
-    :raises TypeError: incorrect type of path variable
+    :param chroot_exe: chroot executable
+    :type chroot_exe: str
+    :raises TypeError: incorrect type of path or chroot_exe variable
 
     .. versionadded:: 4.1.0
     """
 
-    __slots__ = ("_chroot_status", "_conn", "_path")
+    __slots__ = ("_chroot_status", "_chroot_exe_status", "_conn", "_path", "_exe")
 
-    def __init__(self, conn: ExecHelper, path: ChRootPathSetT = None) -> None:
+    def __init__(self, conn: ExecHelper, path: ChRootPathSetT = None, chroot_exe: str = "chroot") -> None:
         """Context manager for call commands with sudo.
 
-        :raises TypeError: incorrect type of path variable
+        :raises TypeError: incorrect type of path or chroot_exe variable
         """
         self._conn: ExecHelper = conn
         self._chroot_status: str | None = conn._chroot_path
+        self._chroot_exe_status: str = conn._chroot_exe
         if path is None or isinstance(path, str):
             self._path: str | None = path
         elif isinstance(path, pathlib.Path):
             self._path = path.as_posix()  # get absolute path
         else:
             raise TypeError(f"path={path!r} is not instance of {ChRootPathSetT}")
+        if isinstance(chroot_exe, str):
+            self._exe: str = chroot_exe
+        else:
+            raise TypeError(f"chroot_exe={chroot_exe!r} is not instance of str")
 
     def __enter__(self) -> None:
         self._conn.__enter__()
         self._chroot_status = self._conn._chroot_path
         self._conn._chroot_path = self._path
+        self._chroot_exe_status = self._conn._chroot_exe
+        self._conn._chroot_exe = self._exe
 
     def __exit__(
         self,
@@ -205,6 +214,7 @@ def __exit__(
         exc_tb: TracebackType | None,
     ) -> None:
         self._conn._chroot_path = self._chroot_status
+        self._conn._chroot_exe = self._chroot_exe_status
         self._conn.__exit__(exc_type, exc_val, exc_tb)
 
 
@@ -224,10 +234,12 @@ class ExecHelper(
     .. versionchanged:: 1.2.0 log_mask_re regex rule for masking cmd
     .. versionchanged:: 1.3.5 make API public to use as interface
     .. versionchanged:: 4.1.0 support chroot
+    .. versionchanged:: 8.1.0 support chroot_exe
     """
 
     __slots__ = (
         "__chroot_path",
+        "__chroot_exe",
         "__context_count",
         "__lock",
         "__logger",
@@ -240,6 +252,7 @@ def __init__(self, log_mask_re: LogMaskReT = None, *, logger: logging.Logger) ->
         self.__logger: logging.Logger = logger
         self.log_mask_re: LogMaskReT = log_mask_re
         self.__chroot_path: str | None = None
+        self.__chroot_exe: str = "chroot"
         self.__context_count = 0
 
     @property
@@ -292,18 +305,51 @@ def _chroot_path(self) -> None:
         """
         self.__chroot_path = None
 
-    def chroot(self, path: ChRootPathSetT) -> _ChRootContext:
+    @property
+    def _chroot_exe(self) -> str:
+        """Exe for chroot
+
+        :rtype: str
+        .. versionadded:: 8.1.0
+        """
+        return self.__chroot_exe
+
+    @_chroot_exe.setter
+    def _chroot_exe(self, new_state: str) -> None:
+        """Executable for chroot if set.
+
+        :param new_state: new exe
+        :type new_state: str
+        :raises TypeError: Not supported exe information
+        .. versionadded:: 8.1.0
+        """
+        if isinstance(new_state, str):
+            self.__chroot_exe = new_state
+        else:
+            raise TypeError(f"chroot_exe is expected to be string, but set {new_state!r}")
+
+    @_chroot_exe.deleter
+    def _chroot_exe(self) -> None:
+        """Restore chroot executable.
+
+        .. versionadded:: 8.1.0
+        """
+        self.__chroot_exe = "chroot"
+
+    def chroot(self, path: ChRootPathSetT, chroot_exe: str = "chroot") -> _ChRootContext:
         """Context manager for changing chroot rules.
 
         :param path: chroot path or none for working without chroot.
         :type path: str | pathlib.Path | None
+        :param chroot_exe: chroot exe
+        :type chroot_exe: str
         :return: context manager with selected chroot state inside
         :rtype: typing.ContextManager
 
         .. Note:: Enter and exit main context manager is produced as well.
         .. versionadded:: 4.1.0
         """
-        return _ChRootContext(conn=self, path=path)
+        return _ChRootContext(conn=self, path=path, chroot_exe=chroot_exe)
 
     @property
     def _context_count(self) -> int:
@@ -348,7 +394,7 @@ def _mask_command(self, cmd: str, log_mask_re: LogMaskReT = None) -> str:
 
         return _helpers.mask_command(cmd.rstrip(), self.log_mask_re, log_mask_re)
 
-    def _prepare_command(self, cmd: str, chroot_path: str | None = None) -> str:
+    def _prepare_command(self, cmd: str, chroot_path: str | None = None, chroot_exe: str = "chroot") -> str:
         """Prepare command: cower chroot and other cases.
 
         :param cmd: main command
@@ -358,7 +404,7 @@ def _prepare_command(self, cmd: str, chroot_path: str | None = None) -> str:
         :return: final command, includes chroot, if required
         :rtype: str
         """
-        return _helpers.chroot_command(cmd, chroot_path=chroot_path or self._chroot_path)
+        return _helpers.chroot_command(cmd, chroot_path=chroot_path or self._chroot_path, chroot_exe=chroot_exe)
 
     @abc.abstractmethod
     def _exec_command(
@@ -427,6 +473,7 @@ def open_execute_context(
         open_stdout: bool = True,
         open_stderr: bool = True,
         chroot_path: str | None = None,
+        chroot_exe: str = "chroot",
         **kwargs: typing.Any,
     ) -> ExecuteContext:
         """Get execution context manager.
@@ -441,6 +488,8 @@ def open_execute_context(
         :type open_stderr: bool
         :param chroot_path: chroot path override
         :type chroot_path: str | None
+        :param chroot_exe: chroot exe override
+        :type chroot_exe: str
         :param kwargs: additional parameters for call.
         :type kwargs: typing.Any
 
@@ -460,6 +509,7 @@ def execute(
         open_stderr: bool = True,
         log_stderr: bool = True,
         chroot_path: str | None = None,
+        chroot_exe: str = "chroot",
         **kwargs: typing.Any,
     ) -> exec_result.ExecResult:
         """Execute command and wait for return code.
@@ -485,6 +535,8 @@ def execute(
         :type log_stderr: bool
         :param chroot_path: chroot path override
         :type chroot_path: str | None
+        :param chroot_exe: chroot exe override
+        :type chroot_exe: str
         :param kwargs: additional parameters for call.
         :type kwargs: typing.Any
         :return: Execution result
@@ -495,6 +547,7 @@ def execute(
         .. versionchanged:: 2.1.0 Allow parallel calls
         .. versionchanged:: 7.0.0 Allow command as list of arguments. Command will be joined with components escaping.
         .. versionchanged:: 8.0.0 chroot path exposed.
+        .. versionchanged:: 8.1.0 chroot_exe added.
         """
         log_level: int = logging.INFO if verbose else logging.DEBUG
         cmd = _helpers.cmd_to_string(command)
@@ -511,6 +564,7 @@ def execute(
             open_stdout=open_stdout,
             open_stderr=open_stderr,
             chroot_path=chroot_path,
+            chroot_exe=chroot_exe,
             **kwargs,
         ) as async_result:
             result: exec_result.ExecResult = self._exec_command(
@@ -540,6 +594,7 @@ def __call__(
         open_stderr: bool = True,
         log_stderr: bool = True,
         chroot_path: str | None = None,
+        chroot_exe: str = "chroot",
         **kwargs: typing.Any,
     ) -> exec_result.ExecResult:
         """Execute command and wait for return code.
@@ -565,6 +620,8 @@ def __call__(
         :type log_stderr: bool
         :param chroot_path: chroot path override
         :type chroot_path: str | None
+        :param chroot_exe: chroot exe override
+        :type chroot_exe: str
         :param kwargs: additional parameters for call.
         :type kwargs: typing.Any
         :return: Execution result
@@ -584,6 +641,7 @@ def __call__(
             open_stderr=open_stderr,
             log_stderr=log_stderr,
             chroot_path=chroot_path,
+            chroot_exe=chroot_exe,
             **kwargs,
         )