diff --git a/doc/source/SSHClient.rst b/doc/source/SSHClient.rst
index c3fbce4..9268fc0 100644
--- a/doc/source/SSHClient.rst
+++ b/doc/source/SSHClient.rst
@@ -10,7 +10,7 @@ API: SSHClient and SSHAuth.
 
     SSHClient helper.
 
-    .. py:method:: __init__(host, port=22, username=None, password=None, private_keys=None, auth=None, )
+    .. py:method:: __init__(host, port=22, username=None, password=None, private_keys=None, auth=None, *, chroot_path=None)
 
         :param host: remote hostname
         :type host: ``str``
@@ -26,6 +26,8 @@ API: SSHClient and SSHAuth.
         :type auth: typing.Optional[SSHAuth]
         :param verbose: show additional error/warning messages
         :type verbose: bool
+        :param chroot_path: chroot path (use chroot if set)
+        :type chroot_path: typing.Optional[str]
 
     .. note:: auth has priority over username/password/private_keys
 
@@ -66,6 +68,11 @@ API: SSHClient and SSHAuth.
         ``bool``
         Paramiko status: ready to use|reconnect required
 
+    .. py:attribute:: chroot_path
+
+        ``typing.Optional[str]``
+        Path for chroot if set.
+
     .. py:attribute:: sudo_mode
 
         ``bool``
@@ -102,6 +109,18 @@ API: SSHClient and SSHAuth.
         .. versionchanged:: 1.1.0 release lock on exit
         .. versionchanged:: 1.2.1 disconnect enforced on close only not in keepalive mode
 
+    .. py:method:: chroot(path)
+
+        Context manager for changing chroot rules.
+
+        :param path: chroot path or none for working without chroot.
+        :type path: typing.Optional[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
+
     .. py:method:: sudo(enforce=None)
 
         Context manager getter for sudo operation
@@ -121,7 +140,7 @@ API: SSHClient and SSHAuth.
         .. Note:: Enter and exit ssh context manager is produced as well.
         .. versionadded:: 1.2.1
 
-    .. py:method:: execute_async(command, stdin=None, open_stdout=True, open_stderr=True, verbose=False, log_mask_re=None, *, get_pty=False, width=80, height=24, **kwargs)
+    .. py:method:: execute_async(command, stdin=None, open_stdout=True, open_stderr=True, verbose=False, log_mask_re=None, *, chroot_path=None, get_pty=False, width=80, height=24, **kwargs)
 
         Execute command in async mode and return channel with IO objects.
 
@@ -137,7 +156,9 @@ API: SSHClient and SSHAuth.
         :type verbose: bool
         :param log_mask_re: regex lookup rule to mask command for logger.
                             all MATCHED groups will be replaced by '<*masked*>'
-        :type log_mask_re: typing.Optional[str]
+        :type log_mask_re: ``typing.Optional[str]``
+        :param chroot_path: chroot path override
+        :type chroot_path: ``typing.Optional[str]``
         :param get_pty: Get PTY for connection
         :type get_pty: bool
         :param width: PTY width
diff --git a/doc/source/Subprocess.rst b/doc/source/Subprocess.rst
index 7839a1e..d52604c 100644
--- a/doc/source/Subprocess.rst
+++ b/doc/source/Subprocess.rst
@@ -8,7 +8,7 @@ API: Subprocess
 
 .. py:class:: Subprocess()
 
-    .. py:method:: __init__(logger, log_mask_re=None, *, logger=logging.getLogger("exec_helpers.subprocess_runner"))
+    .. py:method:: __init__(logger, log_mask_re=None, *, logger=logging.getLogger("exec_helpers.subprocess_runner"), chroot_path=None)
 
         ExecHelper global API.
 
@@ -16,10 +16,13 @@ API: Subprocess
         :type log_mask_re: typing.Optional[str]
         :param logger: logger instance to use
         :type logger: logging.Logger
+        :param chroot_path: chroot path (use chroot if set)
+        :type chroot_path: typing.Optional[str]
 
         .. versionchanged:: 1.2.0 log_mask_re regex rule for masking cmd
         .. versionchanged:: 3.1.0 Not singleton anymore. Only lock is shared between all instances.
         .. versionchanged:: 3.2.0 Logger can be enforced.
+        .. versionchanged:: 4.1.0 support chroot
 
     .. py:attribute:: log_mask_re
 
@@ -43,7 +46,24 @@ API: Subprocess
 
         .. versionchanged:: 1.1.0 release lock on exit
 
-    .. py:method:: execute_async(command, stdin=None, open_stdout=True, open_stderr=True, verbose=False, log_mask_re=None, *, cwd=None, env=None, **kwargs)
+    .. py:attribute:: chroot_path
+
+        ``typing.Optional[str]``
+        Path for chroot if set.
+
+    .. py:method:: chroot(path)
+
+        Context manager for changing chroot rules.
+
+        :param path: chroot path or none for working without chroot.
+        :type path: typing.Optional[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
+
+    .. py:method:: execute_async(command, stdin=None, open_stdout=True, open_stderr=True, verbose=False, log_mask_re=None, *, chroot_path=None, cwd=None, env=None, **kwargs)
 
         Execute command in async mode and return Popen with IO objects.
 
@@ -60,6 +80,8 @@ API: Subprocess
         :param log_mask_re: regex lookup rule to mask command for logger.
                             all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: ``typing.Optional[str]``
+        :param chroot_path: chroot path override
+        :type chroot_path: ``typing.Optional[str]``
         :param cwd: Sets the current directory before the child is executed.
         :type cwd: typing.Optional[typing.Union[str, bytes]]
         :param env: Defines the environment variables for the new process.
diff --git a/exec_helpers/_ssh_client_base.py b/exec_helpers/_ssh_client_base.py
index aea500f..aca7137 100644
--- a/exec_helpers/_ssh_client_base.py
+++ b/exec_helpers/_ssh_client_base.py
@@ -128,6 +128,8 @@ def __call__(  # type: ignore
         private_keys: typing.Optional[typing.Iterable[paramiko.RSAKey]] = None,
         auth: typing.Optional[ssh_auth.SSHAuth] = None,
         verbose: bool = True,
+        *,
+        chroot_path: typing.Optional[str] = None,
     ) -> "SSHClientBase":
         """Main memorize method: check for cached instance and return it. API follows target __init__.
 
@@ -145,10 +147,12 @@ def __call__(  # type: ignore
         :type auth: typing.Optional[ssh_auth.SSHAuth]
         :param verbose: show additional error/warning messages
         :type verbose: bool
+        :param chroot_path: chroot path (use chroot if set)
+        :type chroot_path: typing.Optional[str]
         :return: SSH client instance
         :rtype: SSHClientBase
         """
-        if (host, port) in cls.__cache:
+        if (host, port) in cls.__cache and not chroot_path:  # chrooted connections are not memorized
             key = host, port
             if auth is None:
                 auth = ssh_auth.SSHAuth(username=username, password=password, keys=private_keys)
@@ -176,8 +180,10 @@ def __call__(  # type: ignore
             private_keys=private_keys,
             auth=auth,
             verbose=verbose,
+            chroot_path=chroot_path,
         )
-        cls.__cache[(ssh.hostname, ssh.port)] = ssh
+        if not chroot_path:
+            cls.__cache[(ssh.hostname, ssh.port)] = ssh
         return ssh
 
     @classmethod
@@ -188,7 +194,6 @@ def clear_cache(mcs: typing.Type["_MemorizedSSH"]) -> None:
         """
         n_count = 3
         # PY3: cache, ssh, temporary
-        # PY4: cache, values mapping, ssh, temporary
         for ssh in mcs.__cache.values():
             if CPYTHON and sys.getrefcount(ssh) == n_count:  # pragma: no cover
                 ssh.logger.debug("Closing as unused")
@@ -203,63 +208,66 @@ def close_connections(mcs: typing.Type["_MemorizedSSH"]) -> None:
                 ssh.close()  # type: ignore
 
 
-class SSHClientBase(api.ExecHelper, metaclass=_MemorizedSSH):
-    """SSH Client helper."""
+class _SudoContext:
+    """Context manager for call commands with sudo."""
 
-    __slots__ = ("__hostname", "__port", "__auth", "__ssh", "__sftp", "__sudo_mode", "__keepalive_mode", "__verbose")
+    __slots__ = ("__ssh", "__sudo_status", "__enforce")
 
-    class __get_sudo:
-        """Context manager for call commands with sudo."""
+    def __init__(self, ssh: "SSHClientBase", enforce: typing.Optional[bool] = None) -> None:
+        """Context manager for call commands with sudo.
 
-        __slots__ = ("__ssh", "__sudo_status", "__enforce")
+        :param ssh: connection instance
+        :type ssh: SSHClientBase
+        :param enforce: sudo mode for context manager
+        :type enforce: typing.Optional[bool]
+        """
+        self.__ssh: "SSHClientBase" = ssh
+        self.__sudo_status: bool = ssh.sudo_mode
+        self.__enforce: typing.Optional[bool] = enforce
 
-        def __init__(self, ssh: "SSHClientBase", enforce: typing.Optional[bool] = None) -> None:
-            """Context manager for call commands with sudo.
+    def __enter__(self) -> None:
+        self.__sudo_status = self.__ssh.sudo_mode
+        if self.__enforce is not None:
+            self.__ssh.sudo_mode = self.__enforce
 
-            :param ssh: connection instance
-            :type ssh: SSHClientBase
-            :param enforce: sudo mode for context manager
-            :type enforce: typing.Optional[bool]
-            """
-            self.__ssh: "SSHClientBase" = ssh
-            self.__sudo_status: bool = ssh.sudo_mode
-            self.__enforce: typing.Optional[bool] = enforce
+    def __exit__(self, exc_type: typing.Any, exc_val: typing.Any, exc_tb: typing.Any) -> None:
+        self.__ssh.sudo_mode = self.__sudo_status
 
-        def __enter__(self) -> None:
-            self.__sudo_status = self.__ssh.sudo_mode
-            if self.__enforce is not None:
-                self.__ssh.sudo_mode = self.__enforce
 
-        def __exit__(self, exc_type: typing.Any, exc_val: typing.Any, exc_tb: typing.Any) -> None:
-            self.__ssh.sudo_mode = self.__sudo_status
+class _KeepAliveContext:
+    """Context manager for keepalive management."""
 
-    class __get_keepalive:
-        """Context manager for keepalive management."""
+    __slots__ = ("__ssh", "__keepalive_status", "__enforce")
 
-        __slots__ = ("__ssh", "__keepalive_status", "__enforce")
+    def __init__(self, ssh: "SSHClientBase", enforce: bool = True) -> None:
+        """Context manager for keepalive management.
 
-        def __init__(self, ssh: "SSHClientBase", enforce: bool = True) -> None:
-            """Context manager for keepalive management.
+        :param ssh: connection instance
+        :type ssh: SSHClientBase
+        :param enforce: keepalive mode for context manager
+        :type enforce: bool
+        :param enforce: Keep connection alive after context manager exit
+        """
+        self.__ssh: "SSHClientBase" = ssh
+        self.__keepalive_status: bool = ssh.keepalive_mode
+        self.__enforce: typing.Optional[bool] = enforce
 
-            :param ssh: connection instance
-            :type ssh: SSHClientBase
-            :param enforce: keepalive mode for context manager
-            :type enforce: bool
-            :param enforce: Keep connection alive after context manager exit
-            """
-            self.__ssh: "SSHClientBase" = ssh
-            self.__keepalive_status: bool = ssh.keepalive_mode
-            self.__enforce: typing.Optional[bool] = enforce
+    def __enter__(self) -> None:
+        self.__ssh.__enter__()
+        self.__keepalive_status = self.__ssh.keepalive_mode
+        if self.__enforce is not None:
+            self.__ssh.keepalive_mode = self.__enforce
 
-        def __enter__(self) -> None:
-            self.__keepalive_status = self.__ssh.keepalive_mode
-            if self.__enforce is not None:
-                self.__ssh.keepalive_mode = self.__enforce
-            self.__ssh.__enter__()
+    def __exit__(self, exc_type: typing.Any, exc_val: typing.Any, exc_tb: typing.Any) -> None:
+        # Exit before releasing!
+        self.__ssh.__exit__(exc_type=exc_type, exc_val=exc_val, exc_tb=exc_tb)  # type: ignore
+        self.__ssh.keepalive_mode = self.__keepalive_status
+
+
+class SSHClientBase(api.ExecHelper, metaclass=_MemorizedSSH):
+    """SSH Client helper."""
 
-        def __exit__(self, exc_type: typing.Any, exc_val: typing.Any, exc_tb: typing.Any) -> None:
-            self.__ssh.__exit__(exc_type=exc_type, exc_val=exc_val, exc_tb=exc_tb)  # type: ignore
-            self.__ssh.keepalive_mode = self.__keepalive_status
+    __slots__ = ("__hostname", "__port", "__auth", "__ssh", "__sftp", "__sudo_mode", "__keepalive_mode", "__verbose")
 
     def __hash__(self) -> int:
         """Hash for usage as dict keys."""
@@ -274,6 +282,8 @@ def __init__(
         private_keys: typing.Optional[typing.Iterable[paramiko.RSAKey]] = None,
         auth: typing.Optional[ssh_auth.SSHAuth] = None,
         verbose: bool = True,
+        *,
+        chroot_path: typing.Optional[str] = None,
     ) -> None:
         """Main SSH Client helper.
 
@@ -291,11 +301,13 @@ def __init__(
         :type auth: typing.Optional[ssh_auth.SSHAuth]
         :param verbose: show additional error/warning messages
         :type verbose: bool
+        :param chroot_path: chroot path (use chroot if set)
+        :type chroot_path: typing.Optional[str]
 
         .. note:: auth has priority over username/password/private_keys
         """
         super(SSHClientBase, self).__init__(
-            logger=logging.getLogger(self.__class__.__name__).getChild(f"{host}:{port}")
+            logger=logging.getLogger(self.__class__.__name__).getChild(f"{host}:{port}"), chroot_path=chroot_path
         )
 
         self.__hostname: str = host
@@ -510,7 +522,7 @@ def sudo(self, enforce: typing.Optional[bool] = None) -> "typing.ContextManager[
         :return: context manager with selected sudo state inside
         :rtype: typing.ContextManager
         """
-        return self.__get_sudo(ssh=self, enforce=enforce)
+        return _SudoContext(ssh=self, enforce=enforce)
 
     def keepalive(self, enforce: bool = True) -> "typing.ContextManager[None]":
         """Call contextmanager with keepalive mode change.
@@ -523,7 +535,7 @@ def keepalive(self, enforce: bool = True) -> "typing.ContextManager[None]":
         .. Note:: Enter and exit ssh context manager is produced as well.
         .. versionadded:: 1.2.1
         """
-        return self.__get_keepalive(ssh=self, enforce=enforce)
+        return _KeepAliveContext(ssh=self, enforce=enforce)
 
     # noinspection PyMethodOverriding
     def execute_async(  # pylint: disable=arguments-differ
@@ -535,6 +547,7 @@ def execute_async(  # pylint: disable=arguments-differ
         verbose: bool = False,
         log_mask_re: typing.Optional[str] = None,
         *,
+        chroot_path: typing.Optional[str] = None,
         get_pty: bool = False,
         width: int = 80,
         height: int = 24,
@@ -555,6 +568,8 @@ def execute_async(  # pylint: disable=arguments-differ
         :param log_mask_re: regex lookup rule to mask command for logger.
                             all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: typing.Optional[str]
+        :param chroot_path: chroot path override
+        :type chroot_path: typing.Optional[str]
         :param get_pty: Get PTY for connection
         :type get_pty: bool
         :param width: PTY width
@@ -580,6 +595,7 @@ def execute_async(  # pylint: disable=arguments-differ
         .. versionchanged:: 1.2.0 get_pty moved to `**kwargs`
         .. versionchanged:: 2.1.0 Use typed NamedTuple as result
         .. versionchanged:: 3.2.0 Expose pty options as optional keyword-only arguments
+        .. versionchanged:: 4.1.0 support chroot
         """
         cmd_for_log: str = self._mask_command(cmd=command, log_mask_re=log_mask_re)
 
@@ -597,7 +613,8 @@ def execute_async(  # pylint: disable=arguments-differ
         stdout: paramiko.ChannelFile = chan.makefile("rb")
         stderr: typing.Optional[paramiko.ChannelFile] = chan.makefile_stderr("rb") if open_stderr else None
 
-        cmd = f"{command}\n"
+        cmd = f"{self._prepare_command(cmd=command, chroot_path=chroot_path)}\n"
+
         started = datetime.datetime.utcnow()
         if self.sudo_mode:
             encoded_cmd = base64.b64encode(cmd.encode("utf-8")).decode("utf-8")
diff --git a/exec_helpers/api.py b/exec_helpers/api.py
index 47b0023..7dedb5d 100644
--- a/exec_helpers/api.py
+++ b/exec_helpers/api.py
@@ -46,12 +46,48 @@
 )
 
 
+class _ChRootContext:
+    """Context manager for call commands with chroot.
+
+    .. versionadded:: 4.1.0
+    """
+
+    __slots__ = ("__conn", "__chroot_status", "__path")
+
+    def __init__(self, conn: "ExecHelper", path: typing.Optional[str] = None) -> None:
+        """Context manager for call commands with sudo.
+
+        :param conn: connection instance
+        :type conn: ExecHelper
+        :param path: chroot path or None for no chroot
+        :type path: typing.Optional[str]
+        """
+        self.__conn: "ExecHelper" = conn
+        self.__chroot_status: typing.Optional[str] = conn.chroot_path
+        self.__path: typing.Optional[str] = path
+
+    def __enter__(self) -> None:
+        self.__conn.__enter__()
+        self.__chroot_status = self.__conn.chroot_path
+        self.__conn.chroot_path = self.__path
+
+    def __exit__(self, exc_type: typing.Any, exc_val: typing.Any, exc_tb: typing.Any) -> None:
+        self.__conn.chroot_path = self.__chroot_status
+        self.__conn.__exit__(exc_type=exc_type, exc_val=exc_val, exc_tb=exc_tb)  # type: ignore
+
+
 class ExecHelper(metaclass=abc.ABCMeta):
     """ExecHelper global API."""
 
-    __slots__ = ("__lock", "__logger", "log_mask_re")
+    __slots__ = ("__lock", "__logger", "log_mask_re", "__chroot_path")
 
-    def __init__(self, logger: logging.Logger, log_mask_re: typing.Optional[str] = None) -> None:
+    def __init__(
+        self,
+        log_mask_re: typing.Optional[str] = None,
+        *,
+        logger: logging.Logger,
+        chroot_path: typing.Optional[str] = None,
+    ) -> None:
         """Global ExecHelper API.
 
         :param logger: logger instance to use
@@ -59,13 +95,17 @@ def __init__(self, logger: logging.Logger, log_mask_re: typing.Optional[str] = N
         :param log_mask_re: regex lookup rule to mask command for logger.
                             all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: typing.Optional[str]
+        :param chroot_path: chroot path (use chroot if set)
+        :type chroot_path: typing.Optional[str]
 
         .. 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
         """
         self.__lock = threading.RLock()
         self.__logger: logging.Logger = logger
         self.log_mask_re: typing.Optional[str] = log_mask_re
+        self.__chroot_path: typing.Optional[str] = chroot_path
 
     @property
     def logger(self) -> logging.Logger:
@@ -80,6 +120,46 @@ def lock(self) -> threading.RLock:
         """
         return self.__lock
 
+    @property
+    def chroot_path(self) -> typing.Optional[str]:
+        """Path for chroot if set.
+
+        :rtype: typing.Optional[str]
+        .. versionadded:: 4.1.0
+        """
+        return self.__chroot_path
+
+    @chroot_path.setter
+    def chroot_path(self, new_state: typing.Optional[str]) -> None:
+        """Path for chroot if set.
+
+        :param new_state: new path
+        :type new_state: typing.Optional[str]
+        .. versionadded:: 4.1.0
+        """
+        self.__chroot_path = new_state
+
+    @chroot_path.deleter
+    def chroot_path(self) -> None:
+        """Remove Path for chroot.
+
+        .. versionadded:: 4.1.0
+        """
+        self.__chroot_path = None
+
+    def chroot(self, path: typing.Union[str, None]) -> "typing.ContextManager[None]":
+        """Context manager for changing chroot rules.
+
+        :param path: chroot path or none for working without chroot.
+        :type path: typing.Optional[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)
+
     def __enter__(self) -> "ExecHelper":
         """Get context manager.
 
@@ -136,6 +216,12 @@ def mask(text: str, rules: str) -> str:
 
         return result
 
+    def _prepare_command(self, cmd: str, chroot_path: typing.Optional[str] = None) -> str:
+        """Prepare command: cower chroot and other cases."""
+        if any((chroot_path, self.chroot_path)):
+            return f"chroot {chroot_path if chroot_path else self.chroot_path} {cmd}"
+        return cmd
+
     @abc.abstractmethod
     def execute_async(
         self,
@@ -145,6 +231,8 @@ def execute_async(
         open_stderr: bool = True,
         verbose: bool = False,
         log_mask_re: typing.Optional[str] = None,
+        *,
+        chroot_path: typing.Optional[str] = None,
         **kwargs: typing.Any,
     ) -> ExecuteAsyncResult:
         """Execute command in async mode and return remote interface with IO objects.
@@ -162,6 +250,8 @@ def execute_async(
         :param log_mask_re: regex lookup rule to mask command for logger.
                             all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: typing.Optional[str]
+        :param chroot_path: chroot path override
+        :type chroot_path: typing.Optional[str]
         :param kwargs: additional parameters for call.
         :type kwargs: typing.Any
         :return: NamedTuple with control interface and file-like objects for STDIN/STDERR/STDOUT
@@ -179,6 +269,7 @@ def execute_async(
         .. versionchanged:: 1.2.0 open_stdout and open_stderr flags
         .. versionchanged:: 1.2.0 stdin data
         .. versionchanged:: 2.1.0 Use typed NamedTuple as result
+        .. versionchanged:: 4.1.0 support chroot
         """
         raise NotImplementedError  # pragma: no cover
 
diff --git a/exec_helpers/async_api/api.py b/exec_helpers/async_api/api.py
index 4772a59..2a40b98 100644
--- a/exec_helpers/async_api/api.py
+++ b/exec_helpers/async_api/api.py
@@ -36,7 +36,13 @@ class ExecHelper(api.ExecHelper, metaclass=abc.ABCMeta):
 
     __slots__ = ("__alock",)
 
-    def __init__(self, logger: logging.Logger, log_mask_re: typing.Optional[str] = None) -> None:
+    def __init__(
+        self,
+        log_mask_re: typing.Optional[str] = None,
+        *,
+        logger: logging.Logger,
+        chroot_path: typing.Optional[str] = None,
+    ) -> None:
         """Subprocess helper with timeouts and lock-free FIFO.
 
         :param logger: logger instance to use
@@ -44,8 +50,10 @@ def __init__(self, logger: logging.Logger, log_mask_re: typing.Optional[str] = N
         :param log_mask_re: regex lookup rule to mask command for logger.
                             all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: typing.Optional[str]
+        :param chroot_path: chroot path (use chroot if set)
+        :type chroot_path: typing.Optional[str]
         """
-        super(ExecHelper, self).__init__(logger=logger, log_mask_re=log_mask_re)
+        super(ExecHelper, self).__init__(logger=logger, log_mask_re=log_mask_re, chroot_path=chroot_path)
         self.__alock: typing.Optional[asyncio.Lock] = None
 
     async def __aenter__(self) -> "ExecHelper":
@@ -104,6 +112,8 @@ async def execute_async(  # type: ignore
         open_stderr: bool = True,
         verbose: bool = False,
         log_mask_re: typing.Optional[str] = None,
+        *,
+        chroot_path: typing.Optional[str] = None,
         **kwargs: typing.Any,
     ) -> api.ExecuteAsyncResult:
         """Execute command in async mode and return Popen with IO objects.
@@ -121,6 +131,8 @@ async def execute_async(  # type: ignore
         :param log_mask_re: regex lookup rule to mask command for logger.
                             all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: typing.Optional[str]
+        :param chroot_path: chroot path override
+        :type chroot_path: typing.Optional[str]
         :param kwargs: additional parameters for call.
         :type kwargs: typing.Any
         :return: Tuple with control interface and file-like objects for STDIN/STDERR/STDOUT
diff --git a/exec_helpers/async_api/subprocess_runner.py b/exec_helpers/async_api/subprocess_runner.py
index 0054c3b..d41895b 100644
--- a/exec_helpers/async_api/subprocess_runner.py
+++ b/exec_helpers/async_api/subprocess_runner.py
@@ -72,6 +72,7 @@ def __init__(
         log_mask_re: typing.Optional[str] = None,
         *,
         logger: logging.Logger = logging.getLogger(__name__),  # noqa: B008
+        chroot_path: typing.Optional[str] = None,
     ) -> None:
         """Subprocess helper with timeouts and lock-free FIFO.
 
@@ -80,11 +81,13 @@ def __init__(
         :type log_mask_re: typing.Optional[str]
         :param logger: logger instance to use
         :type logger: logging.Logger
+        :param chroot_path: chroot path (use chroot if set)
+        :type chroot_path: typing.Optional[str]
 
         .. versionchanged:: 3.1.0 Not singleton anymore. Only lock is shared between all instances.
         .. versionchanged:: 3.2.0 Logger can be enforced.
         """
-        super(Subprocess, self).__init__(logger=logger, log_mask_re=log_mask_re)
+        super(Subprocess, self).__init__(logger=logger, log_mask_re=log_mask_re, chroot_path=chroot_path)
 
     async def _exec_command(  # type: ignore
         self,
@@ -168,6 +171,7 @@ async def execute_async(  # type: ignore
         verbose: bool = False,
         log_mask_re: typing.Optional[str] = None,
         *,
+        chroot_path: typing.Optional[str] = None,
         cwd: typing.Optional[typing.Union[str, bytes]] = None,
         env: typing.Optional[
             typing.Union[typing.Mapping[bytes, typing.Union[bytes, str]], typing.Mapping[str, typing.Union[bytes, str]]]
@@ -189,6 +193,8 @@ async def execute_async(  # type: ignore
         :param log_mask_re: regex lookup rule to mask command for logger.
                             all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: typing.Optional[str]
+        :param chroot_path: chroot path override
+        :type chroot_path: typing.Optional[str]
         :param cwd: Sets the current directory before the child is executed.
         :type cwd: typing.Optional[typing.Union[str, bytes]]
         :param env: Defines the environment variables for the new process.
@@ -217,7 +223,7 @@ async def execute_async(  # type: ignore
         started = datetime.datetime.utcnow()
 
         process: asyncio.subprocess.Process = await asyncio.create_subprocess_shell(  # pylint: disable=no-member
-            cmd=command,
+            cmd=self._prepare_command(cmd=command, chroot_path=chroot_path),
             stdout=asyncio.subprocess.PIPE if open_stdout else asyncio.subprocess.DEVNULL,
             stderr=asyncio.subprocess.PIPE if open_stderr else asyncio.subprocess.DEVNULL,
             stdin=asyncio.subprocess.PIPE,
diff --git a/exec_helpers/subprocess_runner.py b/exec_helpers/subprocess_runner.py
index 999f909..c61fc2a 100644
--- a/exec_helpers/subprocess_runner.py
+++ b/exec_helpers/subprocess_runner.py
@@ -68,6 +68,7 @@ def __init__(
         log_mask_re: typing.Optional[str] = None,
         *,
         logger: logging.Logger = logging.getLogger(__name__),  # noqa: B008
+        chroot_path: typing.Optional[str] = None,
     ) -> None:
         """Subprocess helper with timeouts and lock-free FIFO.
 
@@ -78,12 +79,14 @@ def __init__(
         :type log_mask_re: typing.Optional[str]
         :param logger: logger instance to use
         :type logger: logging.Logger
+        :param chroot_path: chroot path (use chroot if set)
+        :type chroot_path: typing.Optional[str]
 
         .. versionchanged:: 1.2.0 log_mask_re regex rule for masking cmd
         .. versionchanged:: 3.1.0 Not singleton anymore. Only lock is shared between all instances.
         .. versionchanged:: 3.2.0 Logger can be enforced.
         """
-        super(Subprocess, self).__init__(logger=logger, log_mask_re=log_mask_re)
+        super(Subprocess, self).__init__(logger=logger, log_mask_re=log_mask_re, chroot_path=chroot_path)
 
     def _exec_command(  # type: ignore
         self,
@@ -190,6 +193,7 @@ def execute_async(  # pylint: disable=arguments-differ
         verbose: bool = False,
         log_mask_re: typing.Optional[str] = None,
         *,
+        chroot_path: typing.Optional[str] = None,
         cwd: typing.Optional[typing.Union[str, bytes]] = None,
         env: typing.Optional[
             typing.Union[typing.Mapping[bytes, typing.Union[bytes, str]], typing.Mapping[str, typing.Union[bytes, str]]]
@@ -211,6 +215,8 @@ def execute_async(  # pylint: disable=arguments-differ
         :param log_mask_re: regex lookup rule to mask command for logger.
                             all MATCHED groups will be replaced by '<*masked*>'
         :type log_mask_re: typing.Optional[str]
+        :param chroot_path: chroot path override
+        :type chroot_path: typing.Optional[str]
         :param cwd: Sets the current directory before the child is executed.
         :type cwd: typing.Optional[typing.Union[str, bytes]]
         :param env: Defines the environment variables for the new process.
@@ -233,6 +239,7 @@ def execute_async(  # pylint: disable=arguments-differ
         .. versionadded:: 1.2.0
         .. versionchanged:: 2.1.0 Use typed NamedTuple as result
         .. versionchanged:: 3.2.0 Expose cwd and env as optional keyword-only arguments
+        .. versionchanged:: 4.1.0 support chroot
         """
         cmd_for_log: str = self._mask_command(cmd=command, log_mask_re=log_mask_re)
 
@@ -243,7 +250,7 @@ def execute_async(  # pylint: disable=arguments-differ
         started = datetime.datetime.utcnow()
 
         process = subprocess.Popen(
-            args=[command],
+            args=[self._prepare_command(cmd=command, chroot_path=chroot_path)],
             stdout=subprocess.PIPE if open_stdout else subprocess.DEVNULL,
             stderr=subprocess.PIPE if open_stderr else subprocess.DEVNULL,
             stdin=subprocess.PIPE,
diff --git a/test/test_ssh_client_execute_async_special.py b/test/test_ssh_client_execute_async_special.py
index c0d24e5..2bcb3b1 100644
--- a/test/test_ssh_client_execute_async_special.py
+++ b/test/test_ssh_client_execute_async_special.py
@@ -228,3 +228,53 @@ def test_010_check_stdin_closed(paramiko_ssh_client, chan_makefile, auto_add_pol
 
     log = get_logger(ssh.__class__.__name__).getChild(f"{host}:{port}")
     log.warning.assert_called_once_with("STDIN Send failed: closed channel")
+
+
+def test_011_execute_async_chroot(ssh, ssh_transport_channel):
+    """Global chroot path."""
+    ssh.chroot_path = "/"
+
+    ssh.execute_async(command)
+    ssh_transport_channel.assert_has_calls(
+        (
+            mock.call.makefile_stderr("rb"),
+            mock.call.exec_command(f'chroot {ssh.chroot_path} {command}\n'),
+        )
+    )
+
+
+def test_012_execute_async_chroot_cmd(ssh, ssh_transport_channel):
+    """Command-only chroot path."""
+    ssh.execute_async(command, chroot_path='/')
+    ssh_transport_channel.assert_has_calls(
+        (
+            mock.call.makefile_stderr("rb"),
+            mock.call.exec_command(f'chroot / {command}\n'),
+        )
+    )
+
+
+def test_013_execute_async_chroot_context(ssh, ssh_transport_channel):
+    """Context-managed chroot path."""
+    with ssh.chroot('/'):
+        ssh.execute_async(command)
+    ssh_transport_channel.assert_has_calls(
+        (
+            mock.call.makefile_stderr("rb"),
+            mock.call.exec_command(f'chroot / {command}\n'),
+        )
+    )
+
+
+def test_014_execute_async_no_chroot_context(ssh, ssh_transport_channel):
+    """Context-managed chroot path override."""
+    ssh.chroot_path = "/"
+
+    with ssh.chroot(None):
+        ssh.execute_async(command)
+    ssh_transport_channel.assert_has_calls(
+        (
+            mock.call.makefile_stderr("rb"),
+            mock.call.exec_command(f'{command}\n'),
+        )
+    )