kiwi.chroot_manager

View Source

  1# Copyright (c) 2025 SUSE Software Solutions Germany GmbH.  All rights reserved.
  2#
  3# This file is part of kiwi.
  4#
  5# kiwi is free software: you can redistribute it and/or modify
  6# it under the terms of the GNU General Public License as published by
  7# the Free Software Foundation, either version 3 of the License, or
  8# (at your option) any later version.
  9#
 10# kiwi is distributed in the hope that it will be useful,
 11# but WITHOUT ANY WARRANTY; without even the implied warranty of
 12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 13# GNU General Public License for more details.
 14#
 15# You should have received a copy of the GNU General Public License
 16# along with kiwi.  If not, see <http://www.gnu.org/licenses/>
 17#
 18import os
 19import logging
 20from typing import (
 21    List, Optional, NamedTuple
 22)
 23
 24# project
 25from kiwi.mount_manager import MountManager
 26from kiwi.command import (
 27    Command, CommandT, MutableMapping
 28)
 29from kiwi.exceptions import (
 30    KiwiUmountBusyError
 31)
 32
 33log = logging.getLogger('kiwi')
 34
 35
 36class ChrootMount(NamedTuple):
 37    target: str
 38    source: Optional[str] = None
 39
 40
 41class ChrootManager:
 42    """
 43    **Implements methods for setting and unsetting a chroot environment**
 44
 45    The caller is responsible for cleaning up bind mounts if the ChrootManager
 46    is used as is, without a context.
 47
 48    The class also supports to be used as a context manager, where any bind or kernel
 49    filesystem mount is unmounted once the context manager's with block is left
 50
 51    * :param string root_dir: path to change the root to
 52    * :param list binds: current root paths to bind to the chrooted path
 53    """
 54    def __init__(self, root_dir: str, binds: List[ChrootMount] = []):
 55        self.root_dir = root_dir
 56        self.mounts: List[MountManager] = []
 57        for bind in binds:
 58            self.mounts.append(MountManager(
 59                device=bind.source if bind.source else bind.target,
 60                mountpoint=os.path.normpath(
 61                    os.sep.join([root_dir, bind.target])
 62                )
 63            ))
 64
 65    def __enter__(self) -> "ChrootManager":
 66        try:
 67            self.mount()
 68        except Exception as e:
 69            try:
 70                self.umount()
 71            except Exception:
 72                pass
 73            raise e
 74        return self
 75
 76    def __exit__(self, exc_type, exc_value, traceback) -> None:
 77        self.umount()
 78
 79    def mount(self) -> None:
 80        """
 81        Mounts binds to the chroot path
 82        """
 83        for mnt in self.mounts:
 84            mnt.bind_mount()
 85
 86    def umount(self) -> None:
 87        """
 88        Unmounts all binds from the chroot path
 89
 90        If any unmount raises a KiwiUmountBusyError this is trapped
 91        and kept until the iteration over all bind mounts is over.
 92        """
 93        errors = []
 94        for mnt in reversed(self.mounts):
 95            try:
 96                mnt.umount()
 97            except KiwiUmountBusyError as e:
 98                errors.append(e)
 99
100        if errors:
101            raise KiwiUmountBusyError(errors)
102
103    def run(
104        self, command: List[str],
105        custom_env: Optional[MutableMapping[str, str]] = None,
106        raise_on_error: bool = True, stderr_to_stdout: bool = False,
107        raise_on_command_not_found: bool = True
108    ) -> Optional[CommandT]:
109        """
110        This is a wrapper for Command.run method but pre-appending the
111        chroot call at the command list
112
113        :param list command: command and arguments
114        :param dict custom_env: custom os.environ
115        :param bool raise_on_error: control error behaviour
116        :param bool stderr_to_stdout: redirects stderr to stdout
117
118        :return:
119            Contains call results in command type
120
121            .. code:: python
122
123                CommandT(output='string', error='string', returncode=int)
124
125        :rtype: CommandT
126        """
127        chroot_cmd = ['chroot', self.root_dir]
128        chroot_cmd = chroot_cmd + command
129        return Command.run(
130            chroot_cmd, custom_env, raise_on_error, stderr_to_stdout,
131            raise_on_command_not_found
132        )

log = <Logger kiwi (DEBUG)>

class ChrootMount(typing.NamedTuple): View Source

37class ChrootMount(NamedTuple):
38    target: str
39    source: Optional[str] = None

ChrootMount(target, source)

ChrootMount(target: str, source: Optional[str] = None)

Create new instance of ChrootMount(target, source)

target: str

Alias for field number 0

source: Optional[str]

Alias for field number 1

class ChrootManager: View Source

 42class ChrootManager:
 43    """
 44    **Implements methods for setting and unsetting a chroot environment**
 45
 46    The caller is responsible for cleaning up bind mounts if the ChrootManager
 47    is used as is, without a context.
 48
 49    The class also supports to be used as a context manager, where any bind or kernel
 50    filesystem mount is unmounted once the context manager's with block is left
 51
 52    * :param string root_dir: path to change the root to
 53    * :param list binds: current root paths to bind to the chrooted path
 54    """
 55    def __init__(self, root_dir: str, binds: List[ChrootMount] = []):
 56        self.root_dir = root_dir
 57        self.mounts: List[MountManager] = []
 58        for bind in binds:
 59            self.mounts.append(MountManager(
 60                device=bind.source if bind.source else bind.target,
 61                mountpoint=os.path.normpath(
 62                    os.sep.join([root_dir, bind.target])
 63                )
 64            ))
 65
 66    def __enter__(self) -> "ChrootManager":
 67        try:
 68            self.mount()
 69        except Exception as e:
 70            try:
 71                self.umount()
 72            except Exception:
 73                pass
 74            raise e
 75        return self
 76
 77    def __exit__(self, exc_type, exc_value, traceback) -> None:
 78        self.umount()
 79
 80    def mount(self) -> None:
 81        """
 82        Mounts binds to the chroot path
 83        """
 84        for mnt in self.mounts:
 85            mnt.bind_mount()
 86
 87    def umount(self) -> None:
 88        """
 89        Unmounts all binds from the chroot path
 90
 91        If any unmount raises a KiwiUmountBusyError this is trapped
 92        and kept until the iteration over all bind mounts is over.
 93        """
 94        errors = []
 95        for mnt in reversed(self.mounts):
 96            try:
 97                mnt.umount()
 98            except KiwiUmountBusyError as e:
 99                errors.append(e)
100
101        if errors:
102            raise KiwiUmountBusyError(errors)
103
104    def run(
105        self, command: List[str],
106        custom_env: Optional[MutableMapping[str, str]] = None,
107        raise_on_error: bool = True, stderr_to_stdout: bool = False,
108        raise_on_command_not_found: bool = True
109    ) -> Optional[CommandT]:
110        """
111        This is a wrapper for Command.run method but pre-appending the
112        chroot call at the command list
113
114        :param list command: command and arguments
115        :param dict custom_env: custom os.environ
116        :param bool raise_on_error: control error behaviour
117        :param bool stderr_to_stdout: redirects stderr to stdout
118
119        :return:
120            Contains call results in command type
121
122            .. code:: python
123
124                CommandT(output='string', error='string', returncode=int)
125
126        :rtype: CommandT
127        """
128        chroot_cmd = ['chroot', self.root_dir]
129        chroot_cmd = chroot_cmd + command
130        return Command.run(
131            chroot_cmd, custom_env, raise_on_error, stderr_to_stdout,
132            raise_on_command_not_found
133        )

Implements methods for setting and unsetting a chroot environment

The caller is responsible for cleaning up bind mounts if the ChrootManager is used as is, without a context.

The class also supports to be used as a context manager, where any bind or kernel filesystem mount is unmounted once the context manager's with block is left

:param string root_dir: path to change the root to
:param list binds: current root paths to bind to the chrooted path

ChrootManager(root_dir: str, binds: List[ChrootMount] = []) View Source

55    def __init__(self, root_dir: str, binds: List[ChrootMount] = []):
56        self.root_dir = root_dir
57        self.mounts: List[MountManager] = []
58        for bind in binds:
59            self.mounts.append(MountManager(
60                device=bind.source if bind.source else bind.target,
61                mountpoint=os.path.normpath(
62                    os.sep.join([root_dir, bind.target])
63                )
64            ))

root_dir

mounts: List[kiwi.mount_manager.MountManager]

def mount(self) -> None: View Source

80    def mount(self) -> None:
81        """
82        Mounts binds to the chroot path
83        """
84        for mnt in self.mounts:
85            mnt.bind_mount()

Mounts binds to the chroot path

def umount(self) -> None: View Source

 87    def umount(self) -> None:
 88        """
 89        Unmounts all binds from the chroot path
 90
 91        If any unmount raises a KiwiUmountBusyError this is trapped
 92        and kept until the iteration over all bind mounts is over.
 93        """
 94        errors = []
 95        for mnt in reversed(self.mounts):
 96            try:
 97                mnt.umount()
 98            except KiwiUmountBusyError as e:
 99                errors.append(e)
100
101        if errors:
102            raise KiwiUmountBusyError(errors)

Unmounts all binds from the chroot path

If any unmount raises a KiwiUmountBusyError this is trapped and kept until the iteration over all bind mounts is over.

def run( self, command: List[str], custom_env: Optional[MutableMapping[str, str]] = None, raise_on_error: bool = True, stderr_to_stdout: bool = False, raise_on_command_not_found: bool = True) -> Optional[kiwi.command.CommandT]: View Source

104    def run(
105        self, command: List[str],
106        custom_env: Optional[MutableMapping[str, str]] = None,
107        raise_on_error: bool = True, stderr_to_stdout: bool = False,
108        raise_on_command_not_found: bool = True
109    ) -> Optional[CommandT]:
110        """
111        This is a wrapper for Command.run method but pre-appending the
112        chroot call at the command list
113
114        :param list command: command and arguments
115        :param dict custom_env: custom os.environ
116        :param bool raise_on_error: control error behaviour
117        :param bool stderr_to_stdout: redirects stderr to stdout
118
119        :return:
120            Contains call results in command type
121
122            .. code:: python
123
124                CommandT(output='string', error='string', returncode=int)
125
126        :rtype: CommandT
127        """
128        chroot_cmd = ['chroot', self.root_dir]
129        chroot_cmd = chroot_cmd + command
130        return Command.run(
131            chroot_cmd, custom_env, raise_on_error, stderr_to_stdout,
132            raise_on_command_not_found
133        )

This is a wrapper for Command.run method but pre-appending the chroot call at the command list

Parameters

list command: command and arguments
dict custom_env: custom os.environ
bool raise_on_error: control error behaviour
bool stderr_to_stdout: redirects stderr to stdout

Returns

Contains call results in command type

.. code:: python

    CommandT(output='string', error='string', returncode=int)