From 82b386532b5f4c7577fb82cb23deb46faaf7c49d Mon Sep 17 00:00:00 2001
From: Adrian Clay Lake <clay.lake@canonical.com>
Date: Thu, 17 Oct 2024 09:32:57 +0200
Subject: [PATCH 1/6] feat: mount the managed instances apt configuration

---
 craft_parts/executor/executor.py        |   3 +
 craft_parts/lifecycle_manager.py        |   2 +
 craft_parts/overlays/chroot.py          | 244 ++++++++++++++++++------
 craft_parts/overlays/errors.py          |  15 ++
 craft_parts/overlays/overlay_manager.py |  22 ++-
 5 files changed, 219 insertions(+), 67 deletions(-)

diff --git a/craft_parts/executor/executor.py b/craft_parts/executor/executor.py
index 55d5ce61..aae91e2e 100644
--- a/craft_parts/executor/executor.py
+++ b/craft_parts/executor/executor.py
@@ -64,6 +64,7 @@ def __init__(
         ignore_patterns: list[str] | None = None,
         base_layer_dir: Path | None = None,
         base_layer_hash: LayerHash | None = None,
+        use_host_sources: bool = False,
     ) -> None:
         self._part_list = sort_parts(part_list)
         self._project_info = project_info
@@ -73,11 +74,13 @@ def __init__(
         self._base_layer_hash = base_layer_hash
         self._handler: dict[str, PartHandler] = {}
         self._ignore_patterns = ignore_patterns
+        self._use_host_sources = use_host_sources
 
         self._overlay_manager = OverlayManager(
             project_info=self._project_info,
             part_list=self._part_list,
             base_layer_dir=base_layer_dir,
+            use_host_sources=use_host_sources,
         )
 
     def prologue(self) -> None:
diff --git a/craft_parts/lifecycle_manager.py b/craft_parts/lifecycle_manager.py
index 547c26c2..53dbb2b2 100644
--- a/craft_parts/lifecycle_manager.py
+++ b/craft_parts/lifecycle_manager.py
@@ -108,6 +108,7 @@ def __init__(  # noqa: PLR0913
         project_vars_part_name: str | None = None,
         project_vars: dict[str, str] | None = None,
         partitions: list[str] | None = None,
+        use_host_sources: bool = False,
         **custom_args: Any,  # custom passthrough args
     ) -> None:
         # pylint: disable=too-many-locals
@@ -193,6 +194,7 @@ def __init__(  # noqa: PLR0913
             track_stage_packages=track_stage_packages,
             base_layer_dir=base_layer_dir,
             base_layer_hash=layer_hash,
+            use_host_sources=use_host_sources,
         )
         self._project_info = project_info
         # pylint: enable=too-many-locals
diff --git a/craft_parts/overlays/chroot.py b/craft_parts/overlays/chroot.py
index 984e1c5e..0bfd1d29 100644
--- a/craft_parts/overlays/chroot.py
+++ b/craft_parts/overlays/chroot.py
@@ -20,10 +20,11 @@
 import multiprocessing
 import os
 import sys
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from multiprocessing.connection import Connection
 from pathlib import Path
-from typing import Any, NamedTuple
+from shutil import copytree, rmtree
+from typing import Any
 
 from craft_parts.utils import os_utils
 
@@ -33,7 +34,11 @@
 
 
 def chroot(
-    path: Path, target: Callable, *args: Any, **kwargs: Any
+    path: Path,
+    target: Callable,
+    use_host_sources: bool = False,  # noqa: FBT001, FBT002
+    args: tuple[Any] = (),  # type: ignore  # noqa: PGH003
+    kwargs: Mapping[str, Any] = {},
 ) -> Any:  # noqa: ANN401
     """Execute a callable in a chroot environment.
 
@@ -50,14 +55,14 @@ def chroot(
         target=_runner, args=(Path(path), child_conn, target, args, kwargs)
     )
     logger.debug("[pid=%d] set up chroot", os.getpid())
-    _setup_chroot(path)
+    _setup_chroot(path, use_host_sources)
     try:
         child.start()
         res, err = parent_conn.recv()
         child.join()
     finally:
         logger.debug("[pid=%d] clean up chroot", os.getpid())
-        _cleanup_chroot(path)
+        _cleanup_chroot(path, use_host_sources)
 
     if isinstance(err, str):
         raise errors.OverlayChrootExecutionError(err)
@@ -86,87 +91,200 @@ def _runner(
     conn.send((res, None))
 
 
-def _setup_chroot(path: Path) -> None:
+def _compare_os_release(host: os_utils.OsRelease, chroot: os_utils.OsRelease):
+    """Compare OsRelease objects from host and chroot for compatibility. See _host_compatible_chroot."""
+    if (host_val := host.id()) != (chroot_val := chroot.id()):
+        errors.IncompatibleChrootError("id", host_val, chroot_val)
+
+    if (host_val := host.version_id()) != (chroot_val := chroot.version_id()):
+        errors.IncompatibleChrootError("version_id", host_val, chroot_val)
+
+
+def _host_compatible_chroot(path: Path) -> bool:
+    """Raise exception if host and chroot are not the same distrobution and release"""
+    # Note: /etc/os-release is symlinked to /usr/lib/os-release
+    # This could cause an issue if /etc/os-release is removed at any point.
+    host_os_release = os_utils.OsRelease()
+    chroot_os_release = os_utils.OsRelease(
+        os_release_file=str(path / "/etc/os-release")
+    )
+    _compare_os_release(host_os_release, chroot_os_release)
+
+
+def _setup_chroot(path: Path, use_host_sources: bool) -> None:
     """Prepare the chroot environment before executing the target function."""
     logger.debug("setup chroot: %r", path)
     if sys.platform == "linux":
-        _setup_chroot_linux(path)
+        # base configuration
+        _setup_chroot_mounts(path, _linux_mounts)
 
+        if use_host_sources:
+            _host_compatible_chroot(path)
 
-def _cleanup_chroot(path: Path) -> None:
+            _setup_chroot_mounts(path, _ubuntu_apt_mounts)
+
+    logger.debug("chroot setup complete")
+
+
+def _cleanup_chroot(path: Path, use_host_sources: bool) -> None:
     """Clean the chroot environment after executing the target function."""
     logger.debug("cleanup chroot: %r", path)
     if sys.platform == "linux":
-        _cleanup_chroot_linux(path)
+        _cleanup_chroot_mounts(path, _linux_mounts)
+
+        if use_host_sources:
+            # Note: no need to check if host is compatible since
+            # we already called _host_compatible_chroot in _setup_chroot
+            _cleanup_chroot_mounts(path, _ubuntu_apt_mounts)
+
+    logger.debug("chroot cleanup complete")
+
+
+class _Mount:
+    def __init__(
+        self, src: str | Path, mountpoint: str | Path, *args, fstype: str | None = None
+    ) -> None:
+        """Mount entry for chroot setup."""
+
+        self.src = Path(src)
+        self.mountpoint = Path(mountpoint)
+        self.args = list(args)
+
+        if fstype is not None:
+            self.args.append(f"-t{fstype}")
+
+    def _mount(self, src: Path, mountpoint: Path, *args) -> None:
+        mountpoint_str = str(mountpoint)
+        src_str = str(src)
+
+        # Only mount if mountpoint exists.
+        pid = os.getpid()
+        if mountpoint.exists():
+            logger.debug("[pid=%d] mount %r on chroot", pid, mountpoint_str)
+            os_utils.mount(src_str, mountpoint_str, *args)
+        else:
+            logger.error("[pid=%d] mountpoint %r does not exist", pid, mountpoint_str)
+
+    @staticmethod
+    def get_abs_path(path: Path, chroot_path: Path):
+        return path / str(chroot_path).lstrip("/")
+
+    def mount_to(self, path: Path, *args) -> None:
+        abs_mountpoint = self.get_abs_path(path, self.mountpoint)
+
+        self._mount(self.src, abs_mountpoint, *self.args, *args)
+
+    def _umount(self, mountpoint: Path, *args) -> None:
+        mountpoint_str = str(mountpoint)
+
+        pid = os.getpid()
+        if mountpoint.exists():
+            logger.debug("[pid=%d] umount: %r", pid, mountpoint_str)
+            os_utils.umount(mountpoint_str, *args)
+        else:
+            logger.warning("[pid=%d] umount: %r not found!", pid, mountpoint_str)
 
+    def unmount_from(self, path: Path, *args) -> None:
+        abs_mountpoint = self.get_abs_path(path, self.mountpoint)
+        self._umount(abs_mountpoint, *args)
 
-class _Mount(NamedTuple):
-    """Mount entry for chroot setup."""
 
-    fstype: str | None
-    src: str
-    mountpoint: str
-    options: list[str] | None
+class _BindMount(_Mount):
+    bind_type = "bind"
+
+    def __init__(self, src: str | Path, mountpoint: str | Path, *args) -> None:
+        super().__init__(src, mountpoint, f"--{self.bind_type}", *args)
+
+    def _mount(self, src: Path, mountpoint: Path, *args) -> None:
+        if src.is_dir():
+            # remove existing content of dir
+            if mountpoint.exists():
+                rmtree(mountpoint)
+
+            # prep mount point
+            mountpoint.mkdir(parents=True, exist_ok=True)
+
+        elif src.is_file():
+            # remove existing file
+            if mountpoint.exists():
+                mountpoint.unlink()
+            else:
+                mountpoint.parent.mkdir(parents=True, exist_ok=True)
+
+            # prep mount point
+            mountpoint.touch()
+        else:
+            raise FileNotFoundError(f"Path not found: {src}")
+
+        super()._mount(src, mountpoint, *args)
+
+
+class _RBindMount(_BindMount):
+    bind_type = "rbind"
+
+    def _umount(self, mountpoint: Path, *args) -> None:
+        super()._umount(mountpoint, "--recursive", "--lazy", *args)
+
+
+class _TempFSClone(_Mount):
+    def __init__(self, src: str, mountpoint: str, *args) -> None:
+        super().__init__(src, mountpoint, *args, fstype="tmpfs")
+
+    def _mount(self, src: Path, mountpoint: Path, *args) -> None:
+        if src.is_dir():
+            # remove existing content of dir
+            if mountpoint.exists():
+                rmtree(mountpoint)
+
+            # prep mount point
+            mountpoint.mkdir(parents=True, exist_ok=True)
+
+        elif src.is_file():
+            raise NotADirectoryError(f"Path is a directory: {src}")
+        else:
+            raise FileNotFoundError(f"Path not found: {src}")
+
+        super()._mount(src, mountpoint, *args)
+
+        copytree(src, mountpoint, dirs_exist_ok=True)
 
 
 # Essential filesystems to mount in order to have basic utilities and
 # name resolution working inside the chroot environment.
+#
+# Some images (such as cloudimgs) symlink ``/etc/resolv.conf`` to
+# ``/run/systemd/resolve/stub-resolv.conf``. We want resolv.conf to be
+# a regular file to bind-mount the host resolver configuration on.
+#
+# There's no need to restore the file to its original condition because
+# this operation happens on a temporary filesystem layer.
 _linux_mounts: list[_Mount] = [
-    _Mount(None, "/etc/resolv.conf", "/etc/resolv.conf", ["--bind"]),
-    _Mount("proc", "proc", "/proc", None),
-    _Mount("sysfs", "sysfs", "/sys", None),
+    _BindMount("/etc/resolv.conf", "/etc/resolv.conf"),
+    _Mount("proc", "/proc", fstype="proc"),
+    _Mount("sysfs", "/sys", fstype="sysfs"),
     # Device nodes require MS_REC to be bind mounted inside a container.
-    _Mount(None, "/dev", "/dev", ["--rbind", "--make-rprivate"]),
+    _RBindMount("/dev", "/dev", "--make-rprivate"),
 ]
 
+# Mounts required to import host's Ubuntu Pro apt configuration to chroot
+# TODO: parameterize this per linux distribution / package manager
+_ubuntu_apt_mounts = [
+    _TempFSClone("/etc/apt", "/etc/apt"),
+    _BindMount("/usr/share/ca-certificates/", "/usr/share/ca-certificates/"),
+    _BindMount("/etc/ssl/certs/", "/etc/ssl/certs/"),
+    _BindMount("/etc/ca-certificates.conf", "/etc/ca-certificates.conf"),
+]
 
-def _setup_chroot_linux(path: Path) -> None:
-    """Linux-specific chroot environment preparation."""
-    # Some images (such as cloudimgs) symlink ``/etc/resolv.conf`` to
-    # ``/run/systemd/resolve/stub-resolv.conf``. We want resolv.conf to be
-    # a regular file to bind-mount the host resolver configuration on.
-    #
-    # There's no need to restore the file to its original condition because
-    # this operation happens on a temporary filesystem layer.
-    resolv_conf = path / "etc/resolv.conf"
-    if resolv_conf.is_symlink():
-        resolv_conf.unlink()
-        resolv_conf.touch()
-    elif not resolv_conf.exists() and resolv_conf.parent.is_dir():
-        resolv_conf.touch()
-
-    pid = os.getpid()
-    for entry in _linux_mounts:
-        args = []
-        if entry.options:
-            args.extend(entry.options)
-        if entry.fstype:
-            args.append(f"-t{entry.fstype}")
-
-        mountpoint = path / entry.mountpoint.lstrip("/")
 
-        # Only mount if mountpoint exists.
-        if mountpoint.exists():
-            logger.debug("[pid=%d] mount %r on chroot", pid, str(mountpoint))
-            os_utils.mount(entry.src, str(mountpoint), *args)
-        else:
-            logger.debug("[pid=%d] mountpoint %r does not exist", pid, str(mountpoint))
+def _setup_chroot_mounts(path: Path, mounts: list[_Mount]) -> None:
+    """Linux-specific chroot environment preparation."""
 
-    logger.debug("chroot setup complete")
+    for entry in mounts:
+        entry.mount_to(path)
 
 
-def _cleanup_chroot_linux(path: Path) -> None:
+def _cleanup_chroot_mounts(path: Path, mounts: list[_Mount]) -> None:
     """Linux-specific chroot environment cleanup."""
-    pid = os.getpid()
-    for entry in reversed(_linux_mounts):
-        mountpoint = path / entry.mountpoint.lstrip("/")
 
-        if mountpoint.exists():
-            logger.debug("[pid=%d] umount: %r", pid, str(mountpoint))
-            if entry.options and "--rbind" in entry.options:
-                # Mount points under /dev may be in use and make the bind mount
-                # unmountable. This may happen in destructive mode depending on
-                # the host environment, so use MNT_DETACH to defer unmounting.
-                os_utils.umount(str(mountpoint), "--recursive", "--lazy")
-            else:
-                os_utils.umount(str(mountpoint))
+    for entry in reversed(mounts):
+        entry.unmount_from(path)
diff --git a/craft_parts/overlays/errors.py b/craft_parts/overlays/errors.py
index 4932d8f9..edb0ab82 100644
--- a/craft_parts/overlays/errors.py
+++ b/craft_parts/overlays/errors.py
@@ -61,3 +61,18 @@ def __init__(self, message: str) -> None:
         brief = f"Overlay environment execution error: {message}"
 
         super().__init__(brief=brief)
+
+
+class IncompatibleChrootError(OverlayError):
+    """Failed to use host package sources because chroot is incompatible.
+
+    :param key: os-release key which was tested.
+    :param host: value from host os-release which was tested.
+    :param chroot: value from chroot os-release which was tested.
+    """
+
+    def __init__(self, key: str, host: str, chroot: str) -> None:
+        self.message = f"key {key} in os-release expected to be {host} found {chroot}"
+        brief = f"Unable to use host sources: {self.message}"
+
+        super().__init__(brief=brief)
diff --git a/craft_parts/overlays/overlay_manager.py b/craft_parts/overlays/overlay_manager.py
index 69ae60a7..4cca833b 100644
--- a/craft_parts/overlays/overlay_manager.py
+++ b/craft_parts/overlays/overlay_manager.py
@@ -39,6 +39,8 @@ class OverlayManager:
     :param part_list: A list of all parts in the project.
     :param base_layer_dir: The directory containing the overlay base, or None
         if the project doesn't use overlay parameters.
+    :param use_host_sources: Configure chroot to use package sources from
+        the the host enviroment.
     """
 
     def __init__(
@@ -47,12 +49,14 @@ def __init__(
         project_info: ProjectInfo,
         part_list: list[Part],
         base_layer_dir: Path | None,
+        use_host_sources: bool = False,
     ) -> None:
         self._project_info = project_info
         self._part_list = part_list
         self._layer_dirs = [p.part_layer_dir for p in part_list]
         self._overlay_fs: OverlayFS | None = None
         self._base_layer_dir = base_layer_dir
+        self._use_host_sources = use_host_sources
 
     @property
     def base_layer_dir(self) -> Path | None:
@@ -128,7 +132,11 @@ def refresh_packages_list(self) -> None:
         mount_dir = self._project_info.overlay_mount_dir
         # Ensure we always run refresh_packages_list by resetting the cache
         packages.Repository.refresh_packages_list.cache_clear()  # type: ignore[attr-defined]
-        chroot.chroot(mount_dir, packages.Repository.refresh_packages_list)
+        chroot.chroot(
+            mount_dir,
+            packages.Repository.refresh_packages_list,
+            use_host_sources=self._use_host_sources,
+        )
 
     def download_packages(self, package_names: list[str]) -> None:
         """Download packages and populate the overlay package cache.
@@ -139,7 +147,12 @@ def download_packages(self, package_names: list[str]) -> None:
             raise RuntimeError("overlay filesystem not mounted")
 
         mount_dir = self._project_info.overlay_mount_dir
-        chroot.chroot(mount_dir, packages.Repository.download_packages, package_names)
+        chroot.chroot(
+            mount_dir,
+            packages.Repository.download_packages,
+            args=(package_names,),
+            use_host_sources=self._use_host_sources,
+        )
 
     def install_packages(self, package_names: list[str]) -> None:
         """Install packages on the overlay area using chroot.
@@ -153,8 +166,9 @@ def install_packages(self, package_names: list[str]) -> None:
         chroot.chroot(
             mount_dir,
             packages.Repository.install_packages,
-            package_names,
-            refresh_package_cache=False,
+            args=(package_names,),
+            use_host_sources=self._use_host_sources,
+            kwargs={"refresh_package_cache": False},
         )
 
 

From 040ee1cbc51dd6b635a4b8c38bdf7083dbdce2d3 Mon Sep 17 00:00:00 2001
From: Adrian Clay Lake <clay.lake@canonical.com>
Date: Thu, 17 Oct 2024 12:11:18 +0200
Subject: [PATCH 2/6] fix: align changes with tests

---
 craft_parts/overlays/chroot.py              | 65 ++++++++++++---------
 tests/unit/overlays/test_chroot.py          | 10 ++--
 tests/unit/overlays/test_overlay_manager.py | 58 +++++++++++++-----
 tests/unit/test_lifecycle_manager.py        |  1 +
 4 files changed, 86 insertions(+), 48 deletions(-)

diff --git a/craft_parts/overlays/chroot.py b/craft_parts/overlays/chroot.py
index 0bfd1d29..17c7425b 100644
--- a/craft_parts/overlays/chroot.py
+++ b/craft_parts/overlays/chroot.py
@@ -153,63 +153,72 @@ def __init__(
         if fstype is not None:
             self.args.append(f"-t{fstype}")
 
-    def _mount(self, src: Path, mountpoint: Path, *args) -> None:
-        mountpoint_str = str(mountpoint)
-        src_str = str(src)
-
-        # Only mount if mountpoint exists.
-        pid = os.getpid()
-        if mountpoint.exists():
-            logger.debug("[pid=%d] mount %r on chroot", pid, mountpoint_str)
-            os_utils.mount(src_str, mountpoint_str, *args)
-        else:
-            logger.error("[pid=%d] mountpoint %r does not exist", pid, mountpoint_str)
-
     @staticmethod
     def get_abs_path(path: Path, chroot_path: Path):
         return path / str(chroot_path).lstrip("/")
 
-    def mount_to(self, path: Path, *args) -> None:
-        abs_mountpoint = self.get_abs_path(path, self.mountpoint)
+    def mountpoint_exists(self, mountpoint: Path):
+        return mountpoint.exists()
+
+    def _mount(self, src: Path, mountpoint: Path, *args: str) -> None:
+        logger.debug("[pid=%d] mount %r on chroot", os.getpid(), src)
+        os_utils.mount(str(src), str(mountpoint), *args)
+
+    def _umount(self, mountpoint: Path, *args: str) -> None:
+        logger.debug("[pid=%d] umount: %r", os.getpid(), mountpoint)
+        os_utils.umount(str(mountpoint), *args)
+
+    def mount_to(self, chroot: Path, *args: str) -> None:
+        abs_mountpoint = self.get_abs_path(chroot, self.mountpoint)
+
+        if not self.mountpoint_exists(abs_mountpoint):
+            logger.warning("[pid=%d] mount: %r not found!", os.getpid(), abs_mountpoint)
+            return
 
         self._mount(self.src, abs_mountpoint, *self.args, *args)
 
-    def _umount(self, mountpoint: Path, *args) -> None:
-        mountpoint_str = str(mountpoint)
+    def unmount_from(self, chroot: Path, *args: str) -> None:
+        abs_mountpoint = self.get_abs_path(chroot, self.mountpoint)
 
-        pid = os.getpid()
-        if mountpoint.exists():
-            logger.debug("[pid=%d] umount: %r", pid, mountpoint_str)
-            os_utils.umount(mountpoint_str, *args)
-        else:
-            logger.warning("[pid=%d] umount: %r not found!", pid, mountpoint_str)
+        if not self.mountpoint_exists(abs_mountpoint):
+            logger.warning("[pid=%d] umount: %r not found!", os.getpid(), chroot)
+            return
 
-    def unmount_from(self, path: Path, *args) -> None:
-        abs_mountpoint = self.get_abs_path(path, self.mountpoint)
         self._umount(abs_mountpoint, *args)
 
 
 class _BindMount(_Mount):
     bind_type = "bind"
 
-    def __init__(self, src: str | Path, mountpoint: str | Path, *args) -> None:
+    def __init__(
+        self,
+        src: str | Path,
+        mountpoint: str | Path,
+        *args: str,
+    ) -> None:
         super().__init__(src, mountpoint, f"--{self.bind_type}", *args)
 
-    def _mount(self, src: Path, mountpoint: Path, *args) -> None:
+    def mountpoint_exists(self, mountpoint: Path):
+        if self.src.exists() and self.src.is_file():
+            return mountpoint.parent.exists()
+
+        return mountpoint.exists()
+
+    def _mount(self, src: Path, mountpoint: Path, *args: str) -> None:
         if src.is_dir():
             # remove existing content of dir
             if mountpoint.exists():
                 rmtree(mountpoint)
 
             # prep mount point
-            mountpoint.mkdir(parents=True, exist_ok=True)
+            mountpoint.mkdir(exist_ok=True)
 
         elif src.is_file():
             # remove existing file
             if mountpoint.exists():
                 mountpoint.unlink()
             else:
-                mountpoint.parent.mkdir(parents=True, exist_ok=True)
+                mountpoint.parent.mkdir(exist_ok=True)
 
             # prep mount point
             mountpoint.touch()
diff --git a/tests/unit/overlays/test_chroot.py b/tests/unit/overlays/test_chroot.py
index de9f6626..4bb2df10 100644
--- a/tests/unit/overlays/test_chroot.py
+++ b/tests/unit/overlays/test_chroot.py
@@ -62,7 +62,7 @@ def test_chroot(self, mocker, new_dir, mock_chroot):
         for subdir in ["etc", "proc", "sys", "dev", "dev/shm"]:
             Path(new_root, subdir).mkdir()
 
-        chroot.chroot(new_root, target_func, "content")
+        chroot.chroot(new_root, target_func, args=("content",))
 
         assert Path("dir1/foo.txt").read_text() == "content"
         assert spy_process.mock_calls == [
@@ -95,7 +95,7 @@ def test_chroot_no_mountpoints(self, mocker, new_dir):
         mocker.patch("os.chroot")
 
         Path("dir1").mkdir()
-        chroot.chroot(new_root, target_func, "content")
+        chroot.chroot(new_root, target_func, args=("content",))
 
         assert Path("dir1/foo.txt").read_text() == "content"
         assert spy_process.mock_calls == [
@@ -119,8 +119,8 @@ def test_chroot_symlinked_resolv_conf(self, mocker, new_dir):
 
         Path("dir1").mkdir()
         Path("dir1/etc").mkdir()
-        Path("dir1/etc/resolv.con").symlink_to("whatever")
-        chroot.chroot(new_root, target_func, "content")
+        Path("dir1/etc/resolv.conf").symlink_to("whatever")
+        chroot.chroot(new_root, target_func, args=("content",))
 
         assert Path("dir1/foo.txt").read_text() == "content"
         assert spy_process.mock_calls == [
@@ -148,7 +148,7 @@ def test_chroot_no_resolv_conf(self, mocker, new_dir):
 
         Path("dir1").mkdir()
         Path("dir1/etc").mkdir()
-        chroot.chroot(new_root, target_func, "content")
+        chroot.chroot(new_root, target_func, args=("content",))
 
         assert Path("dir1/foo.txt").read_text() == "content"
         assert spy_process.mock_calls == [
diff --git a/tests/unit/overlays/test_overlay_manager.py b/tests/unit/overlays/test_overlay_manager.py
index a71f1080..7515dbbd 100644
--- a/tests/unit/overlays/test_overlay_manager.py
+++ b/tests/unit/overlays/test_overlay_manager.py
@@ -177,9 +177,13 @@ def test_refresh_packages_list(self, new_dir):
             f"workdir={new_dir}/overlay/work",
         )
         self.mock_chroot.assert_called_once_with(
-            new_dir / "overlay/overlay", self.mock_refresh_packages_list
+            new_dir / "overlay/overlay",
+            self.mock_refresh_packages_list,
+            use_host_sources=False,
+        )
+        self.mock_refresh_packages_list.assert_called_once_with(
+            use_host_sources=False,
         )
-        self.mock_refresh_packages_list.assert_called_once_with()
 
     def test_download_packages(self, mocker, new_dir):
         mock_download_packages = mocker.patch(
@@ -196,9 +200,15 @@ def test_download_packages(self, mocker, new_dir):
             f"workdir={new_dir}/overlay/work",
         )
         self.mock_chroot.assert_called_once_with(
-            new_dir / "overlay/overlay", mock_download_packages, ["pkg1", "pkg2"]
+            new_dir / "overlay/overlay",
+            mock_download_packages,
+            args=(["pkg1", "pkg2"],),
+            use_host_sources=False,
+        )
+        mock_download_packages.assert_called_once_with(
+            args=(["pkg1", "pkg2"],),
+            use_host_sources=False,
         )
-        mock_download_packages.assert_called_once_with(["pkg1", "pkg2"])
 
     def test_install_packages(self, mocker, new_dir):
         mock_install_packages = mocker.patch(
@@ -218,11 +228,14 @@ def test_install_packages(self, mocker, new_dir):
         self.mock_chroot.assert_called_once_with(
             new_dir / "overlay/overlay",
             mock_install_packages,
-            ["pkg1", "pkg2"],
-            refresh_package_cache=False,
+            args=(["pkg1", "pkg2"],),
+            kwargs={"refresh_package_cache": False},
+            use_host_sources=False,
         )
         mock_install_packages.assert_called_once_with(
-            ["pkg1", "pkg2"], refresh_package_cache=False
+            args=(["pkg1", "pkg2"],),
+            kwargs={"refresh_package_cache": False},
+            use_host_sources=False,
         )
 
     def test_package_cache_mount_refresh(self, new_dir):
@@ -242,9 +255,13 @@ def test_package_cache_mount_refresh(self, new_dir):
             f"workdir={new_dir}/overlay/work",
         )
         self.mock_chroot.assert_called_once_with(
-            new_dir / "overlay/overlay", self.mock_refresh_packages_list
+            new_dir / "overlay/overlay",
+            self.mock_refresh_packages_list,
+            use_host_sources=False,
+        )
+        self.mock_refresh_packages_list.assert_called_once_with(
+            use_host_sources=False,
         )
-        self.mock_refresh_packages_list.assert_called_once_with()
         self.mock_umount.assert_called_once_with(new_dir / "overlay/overlay")
 
     def test_package_cache_mount_download(self, mocker, new_dir):
@@ -267,10 +284,18 @@ def test_package_cache_mount_download(self, mocker, new_dir):
             f"workdir={new_dir}/overlay/work",
         )
         self.mock_chroot.assert_called_once_with(
-            new_dir / "overlay/overlay", mock_download_packages, ["pkg1", "pkg2"]
+            new_dir / "overlay/overlay",
+            mock_download_packages,
+            args=(["pkg1", "pkg2"],),
+            use_host_sources=False,
+        )
+        mock_download_packages.assert_called_once_with(
+            args=(["pkg1", "pkg2"],),
+            use_host_sources=False,
+        )
+        self.mock_umount.assert_called_once_with(
+            new_dir / "overlay/overlay",
         )
-        mock_download_packages.assert_called_once_with(["pkg1", "pkg2"])
-        self.mock_umount.assert_called_once_with(new_dir / "overlay/overlay")
 
     def test_layer_mount_install(self, mocker, new_dir):
         mocker.patch("craft_parts.packages.Repository.download_packages")
@@ -296,10 +321,13 @@ def test_layer_mount_install(self, mocker, new_dir):
         self.mock_chroot.assert_called_once_with(
             new_dir / "overlay/overlay",
             mock_install_packages,
-            ["pkg1", "pkg2"],
-            refresh_package_cache=False,
+            args=(["pkg1", "pkg2"],),
+            kwargs={"refresh_package_cache": False},
+            use_host_sources=False,
         )
         mock_install_packages.assert_called_once_with(
-            ["pkg1", "pkg2"], refresh_package_cache=False
+            args=(["pkg1", "pkg2"],),
+            kwargs={"refresh_package_cache": False},
+            use_host_sources=False,
         )
         self.mock_umount.assert_called_once_with(new_dir / "overlay/overlay")
diff --git a/tests/unit/test_lifecycle_manager.py b/tests/unit/test_lifecycle_manager.py
index 01d0fbfc..b52a8b28 100644
--- a/tests/unit/test_lifecycle_manager.py
+++ b/tests/unit/test_lifecycle_manager.py
@@ -213,6 +213,7 @@ def test_executor_creation(self, new_dir, mocker):
                 track_stage_packages=False,
                 base_layer_dir=None,
                 base_layer_hash=None,
+                use_host_sources=False,
             )
         ]
 

From fbc8ac7d7e30ce0f4339fd437717adf60335a6b3 Mon Sep 17 00:00:00 2001
From: Adrian Clay Lake <clay.lake@canonical.com>
Date: Thu, 17 Oct 2024 13:51:21 +0200
Subject: [PATCH 3/6] fix: ensure mount point exists for binds

---
 craft_parts/overlays/chroot.py | 35 ++++++++++++++++++++++++----------
 1 file changed, 25 insertions(+), 10 deletions(-)

diff --git a/craft_parts/overlays/chroot.py b/craft_parts/overlays/chroot.py
index 17c7425b..2f77bf87 100644
--- a/craft_parts/overlays/chroot.py
+++ b/craft_parts/overlays/chroot.py
@@ -142,13 +142,19 @@ def _cleanup_chroot(path: Path, use_host_sources: bool) -> None:
 
 class _Mount:
     def __init__(
-        self, src: str | Path, mountpoint: str | Path, *args, fstype: str | None = None
+        self,
+        src: str | Path,
+        mountpoint: str | Path,
+        *args,
+        fstype: str | None = None,
+        skip_missing: bool = True,
     ) -> None:
         """Mount entry for chroot setup."""
 
         self.src = Path(src)
         self.mountpoint = Path(mountpoint)
         self.args = list(args)
+        self.skip_missing = skip_missing
 
         if fstype is not None:
             self.args.append(f"-t{fstype}")
@@ -171,7 +177,7 @@ def _umount(self, mountpoint: Path, *args: str) -> None:
     def mount_to(self, chroot: Path, *args: str) -> None:
         abs_mountpoint = self.get_abs_path(chroot, self.mountpoint)
 
-        if not self.mountpoint_exists(abs_mountpoint):
+        if self.skip_missing and not self.mountpoint_exists(abs_mountpoint):
             logger.warning("[pid=%d] mount: %r not found!", os.getpid(), abs_mountpoint)
             return
 
@@ -180,7 +186,7 @@ def mount_to(self, chroot: Path, *args: str) -> None:
     def unmount_from(self, chroot: Path, *args: str) -> None:
         abs_mountpoint = self.get_abs_path(chroot, self.mountpoint)
 
-        if not self.mountpoint_exists(abs_mountpoint):
+        if self.skip_missing and not self.mountpoint_exists(abs_mountpoint):
             logger.warning("[pid=%d] umount: %r not found!", os.getpid(), chroot)
             return
 
@@ -195,8 +201,11 @@ def __init__(
         src: str | Path,
         mountpoint: str | Path,
         *args: str,
+        skip_missing: bool = True,
     ) -> None:
-        super().__init__(src, mountpoint, f"--{self.bind_type}", *args)
+        super().__init__(
+            src, mountpoint, f"--{self.bind_type}", *args, skip_missing=skip_missing
+        )
 
     def mountpoint_exists(self, mountpoint: Path):
         if self.src.exists() and self.src.is_file():
@@ -211,14 +220,14 @@ def _mount(self, src: Path, mountpoint: Path, *args: str) -> None:
                 rmtree(mountpoint)
 
             # prep mount point
-            mountpoint.mkdir(exist_ok=True)
+            mountpoint.mkdir(parents=True, exist_ok=True)
 
         elif src.is_file():
             # remove existing file
             if mountpoint.exists():
                 mountpoint.unlink()
             else:
-                mountpoint.parent.mkdir(exist_ok=True)
+                mountpoint.parent.mkdir(parents=True, exist_ok=True)
 
             # prep mount point
             mountpoint.touch()
@@ -237,7 +246,7 @@ def _umount(self, mountpoint: Path, *args) -> None:
 
 class _TempFSClone(_Mount):
     def __init__(self, src: str, mountpoint: str, *args) -> None:
-        super().__init__(src, mountpoint, *args, fstype="tmpfs")
+        super().__init__(src, mountpoint, *args, fstype="tmpfs", skip_missing=False)
 
     def _mount(self, src: Path, mountpoint: Path, *args) -> None:
         if src.is_dir():
@@ -279,9 +288,15 @@ def _mount(self, src: Path, mountpoint: Path, *args) -> None:
 # TODO: parameterize this per linux distribution / package manager
 _ubuntu_apt_mounts = [
     _TempFSClone("/etc/apt", "/etc/apt"),
-    _BindMount("/usr/share/ca-certificates/", "/usr/share/ca-certificates/"),
-    _BindMount("/etc/ssl/certs/", "/etc/ssl/certs/"),
-    _BindMount("/etc/ca-certificates.conf", "/etc/ca-certificates.conf"),
+    _BindMount(
+        "/usr/share/ca-certificates/",
+        "/usr/share/ca-certificates/",
+        skip_missing=False,
+    ),
+    _BindMount("/etc/ssl/certs/", "/etc/ssl/certs/", skip_missing=False),
+    _BindMount(
+        "/etc/ca-certificates.conf", "/etc/ca-certificates.conf", skip_missing=False
+    ),
 ]
 
 

From 100bdfc03b6cc493b59396ebecb0ed2638c4adf2 Mon Sep 17 00:00:00 2001
From: Adrian Clay Lake <clay.lake@canonical.com>
Date: Tue, 22 Oct 2024 21:32:41 +0200
Subject: [PATCH 4/6] fix: linter spelling warnings

---
 craft_parts/overlays/chroot.py          | 2 +-
 craft_parts/overlays/overlay_manager.py | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/craft_parts/overlays/chroot.py b/craft_parts/overlays/chroot.py
index 2f77bf87..ec00b941 100644
--- a/craft_parts/overlays/chroot.py
+++ b/craft_parts/overlays/chroot.py
@@ -101,7 +101,7 @@ def _compare_os_release(host: os_utils.OsRelease, chroot: os_utils.OsRelease):
 
 
 def _host_compatible_chroot(path: Path) -> bool:
-    """Raise exception if host and chroot are not the same distrobution and release"""
+    """Raise exception if host and chroot are not the same distribution and release"""
     # Note: /etc/os-release is symlinked to /usr/lib/os-release
     # This could cause an issue if /etc/os-release is removed at any point.
     host_os_release = os_utils.OsRelease()
diff --git a/craft_parts/overlays/overlay_manager.py b/craft_parts/overlays/overlay_manager.py
index 4cca833b..ec4150a7 100644
--- a/craft_parts/overlays/overlay_manager.py
+++ b/craft_parts/overlays/overlay_manager.py
@@ -40,7 +40,7 @@ class OverlayManager:
     :param base_layer_dir: The directory containing the overlay base, or None
         if the project doesn't use overlay parameters.
     :param use_host_sources: Configure chroot to use package sources from
-        the the host enviroment.
+        the the host environment.
     """
 
     def __init__(

From 250e56fcfbc8d45ddc3d2311cdbcd6f0767638d9 Mon Sep 17 00:00:00 2001
From: Adrian Clay Lake <clay.lake@canonical.com>
Date: Wed, 23 Oct 2024 17:17:57 +0200
Subject: [PATCH 5/6] fix: refactor and do not rmtree on mounting dir

---
 craft_parts/overlays/chroot.py | 136 +++++++++++++++++----------------
 1 file changed, 71 insertions(+), 65 deletions(-)

diff --git a/craft_parts/overlays/chroot.py b/craft_parts/overlays/chroot.py
index ec00b941..17c65fa5 100644
--- a/craft_parts/overlays/chroot.py
+++ b/craft_parts/overlays/chroot.py
@@ -144,53 +144,67 @@ class _Mount:
     def __init__(
         self,
         src: str | Path,
-        mountpoint: str | Path,
+        dst: str | Path,
         *args,
         fstype: str | None = None,
         skip_missing: bool = True,
     ) -> None:
-        """Mount entry for chroot setup."""
+        """Management class for chroot mounts."""
 
         self.src = Path(src)
-        self.mountpoint = Path(mountpoint)
+        self.dst = Path(dst)
         self.args = list(args)
         self.skip_missing = skip_missing
 
         if fstype is not None:
             self.args.append(f"-t{fstype}")
 
-    @staticmethod
-    def get_abs_path(path: Path, chroot_path: Path):
+        logger.debug("[pid=%d] Mount Manager %s", os.getpid(), self)
+
+    def _mount(self, src: Path, chroot: Path, *args: str) -> None:
+        abs_dst = self.get_abs_path(chroot, self.dst)
+        os_utils.mount(str(src), str(abs_dst), *args)
+
+    def _umount(self, chroot: Path, *args: str) -> None:
+        abs_dst = self.get_abs_path(chroot, self.dst)
+        os_utils.umount(str(abs_dst), *args)
+
+    def get_abs_path(self, path: Path, chroot_path: Path) -> Path:
+        """Make `chroot_path` relative to host `path`"""
         return path / str(chroot_path).lstrip("/")
 
-    def mountpoint_exists(self, mountpoint: Path):
-        return mountpoint.exists()
+    def dst_exists(self, chroot: Path):
+        abs_dst = self.get_abs_path(chroot, self.dst)
+        return abs_dst.is_symlink() or abs_dst.exists()
 
-    def _mount(self, src: Path, mountpoint: Path, *args: str) -> None:
-        logger.debug("[pid=%d] mount %r on chroot", os.getpid(), src)
-        os_utils.mount(str(src), str(mountpoint), *args)
+    def remove_dst(self, chroot: Path):
+        pass
 
-    def _umount(self, mountpoint: Path, *args: str) -> None:
-        logger.debug("[pid=%d] umount: %r", os.getpid(), mountpoint)
-        os_utils.umount(str(mountpoint), *args)
+    def create_dst(self, chroot: Path):
+        abs_dst = self.get_abs_path(chroot, self.dst)
+        abs_dst.parent.mkdir(parents=True, exist_ok=True)
 
     def mount_to(self, chroot: Path, *args: str) -> None:
-        abs_mountpoint = self.get_abs_path(chroot, self.mountpoint)
-
-        if self.skip_missing and not self.mountpoint_exists(abs_mountpoint):
-            logger.warning("[pid=%d] mount: %r not found!", os.getpid(), abs_mountpoint)
+        logger.debug(f"Mounting {self.dst}")
+        if self.dst_exists(chroot):
+            self.remove_dst(chroot)
+        elif self.skip_missing:
+            abs_dst = self.get_abs_path(chroot, self.dst)
+            logger.warning(
+                "[pid=%d] mount: %r not found. Skipping", os.getpid(), abs_dst
+            )
             return
-
-        self._mount(self.src, abs_mountpoint, *self.args, *args)
+        self.create_dst(chroot)
+        self._mount(self.src, chroot, *self.args, *args)
 
     def unmount_from(self, chroot: Path, *args: str) -> None:
-        abs_mountpoint = self.get_abs_path(chroot, self.mountpoint)
-
-        if self.skip_missing and not self.mountpoint_exists(abs_mountpoint):
-            logger.warning("[pid=%d] umount: %r not found!", os.getpid(), chroot)
+        logger.debug(f"Mounting {self.dst}")
+        if self.skip_missing and not self.dst_exists(chroot):
+            abs_dst = self.get_abs_path(chroot, self.dst)
+            logger.warning("[pid=%d] umount: %r not found!", os.getpid(), abs_dst)
             return
 
-        self._umount(abs_mountpoint, *args)
+        self._umount(chroot, *args)
 
 
 class _BindMount(_Mount):
@@ -199,72 +213,64 @@ class _BindMount(_Mount):
     def __init__(
         self,
         src: str | Path,
-        mountpoint: str | Path,
+        dst: str | Path,
         *args: str,
         skip_missing: bool = True,
     ) -> None:
         super().__init__(
-            src, mountpoint, f"--{self.bind_type}", *args, skip_missing=skip_missing
+            src, dst, f"--{self.bind_type}", *args, skip_missing=skip_missing
         )
 
-    def mountpoint_exists(self, mountpoint: Path):
-        if self.src.exists() and self.src.is_file():
-            return mountpoint.parent.exists()
+    def dst_exists(self, chroot: Path):
+        abs_dst = self.get_abs_path(chroot, self.dst)
+
+        if self.src.is_file():
+            return abs_dst.is_symlink() or abs_dst.exists() or abs_dst.parent.is_dir()
+
+        return abs_dst.is_symlink() or abs_dst.exists()
 
-        return mountpoint.exists()
+    def create_dst(self, chroot: Path):
+        abs_dst = self.get_abs_path(chroot, self.dst)
 
-    def _mount(self, src: Path, mountpoint: Path, *args: str) -> None:
-        if src.is_dir():
-            # remove existing content of dir
-            if mountpoint.exists():
-                rmtree(mountpoint)
+        if self.src.is_dir():
+            abs_dst.mkdir(parents=True, exist_ok=True)
+        elif self.src.is_file():
+            abs_dst.touch()
 
-            # prep mount point
-            mountpoint.mkdir(parents=True, exist_ok=True)
+    def remove_dst(self, chroot: Path):
+        abs_dst = self.get_abs_path(chroot, self.dst)
 
-        elif src.is_file():
-            # remove existing file
-            if mountpoint.exists():
-                mountpoint.unlink()
-            else:
-                mountpoint.parent.mkdir(parents=True, exist_ok=True)
+        if abs_dst.is_symlink() or abs_dst.is_file():
+            abs_dst.unlink()
 
-            # prep mount point
-            mountpoint.touch()
-        else:
+    def _mount(self, src: Path, chroot: Path, *args: str) -> None:
+        if not src.exists():
             raise FileNotFoundError(f"Path not found: {src}")
 
-        super()._mount(src, mountpoint, *args)
+        super()._mount(src, chroot, *args)
 
 
 class _RBindMount(_BindMount):
     bind_type = "rbind"
 
-    def _umount(self, mountpoint: Path, *args) -> None:
-        super()._umount(mountpoint, "--recursive", "--lazy", *args)
+    def _umount(self, chroot: Path, *args) -> None:
+        super()._umount(chroot, "--recursive", "--lazy", *args)
 
 
 class _TempFSClone(_Mount):
-    def __init__(self, src: str, mountpoint: str, *args) -> None:
-        super().__init__(src, mountpoint, *args, fstype="tmpfs", skip_missing=False)
-
-    def _mount(self, src: Path, mountpoint: Path, *args) -> None:
-        if src.is_dir():
-            # remove existing content of dir
-            if mountpoint.exists():
-                rmtree(mountpoint)
-
-            # prep mount point
-            mountpoint.mkdir(parents=True, exist_ok=True)
+    def __init__(self, src: str, dst: str, *args) -> None:
+        super().__init__(src, dst, *args, fstype="tmpfs", skip_missing=False)
 
-        elif src.is_file():
-            raise NotADirectoryError(f"Path is a directory: {src}")
-        else:
+    def _mount(self, src: Path, chroot: Path, *args) -> None:
+        if src.is_file():
+            raise NotADirectoryError(f"Path is a file: {src}")
+        if not src.exists():
             raise FileNotFoundError(f"Path not found: {src}")
 
-        super()._mount(src, mountpoint, *args)
+        super()._mount(src, chroot, *args)
 
-        copytree(src, mountpoint, dirs_exist_ok=True)
+        abs_dst = self.get_abs_path(chroot, self.dst)
+        copytree(src, abs_dst, dirs_exist_ok=True)
 
 
 # Essential filesystems to mount in order to have basic utilities and

From f35f8228e037edc912c5ab12085c6420c5f79798 Mon Sep 17 00:00:00 2001
From: Adrian Clay Lake <clay.lake@canonical.com>
Date: Wed, 23 Oct 2024 20:38:56 +0200
Subject: [PATCH 6/6] refactor: QC and linting

---
 craft_parts/executor/executor.py |  2 +-
 craft_parts/overlays/chroot.py   | 85 ++++++++++++++++++++++----------
 2 files changed, 59 insertions(+), 28 deletions(-)

diff --git a/craft_parts/executor/executor.py b/craft_parts/executor/executor.py
index aae91e2e..aaa79203 100644
--- a/craft_parts/executor/executor.py
+++ b/craft_parts/executor/executor.py
@@ -53,7 +53,7 @@ class Executor:
     :param ignore_patterns: File patterns to ignore when pulling local sources.
     """
 
-    def __init__(
+    def __init__(  # noqa: PLR0913
         self,
         *,
         part_list: list[Part],
diff --git a/craft_parts/overlays/chroot.py b/craft_parts/overlays/chroot.py
index 17c65fa5..cd4c682c 100644
--- a/craft_parts/overlays/chroot.py
+++ b/craft_parts/overlays/chroot.py
@@ -23,7 +23,7 @@
 from collections.abc import Callable, Mapping
 from multiprocessing.connection import Connection
 from pathlib import Path
-from shutil import copytree, rmtree
+from shutil import copytree
 from typing import Any
 
 from craft_parts.utils import os_utils
@@ -91,17 +91,17 @@ def _runner(
     conn.send((res, None))
 
 
-def _compare_os_release(host: os_utils.OsRelease, chroot: os_utils.OsRelease):
+def _compare_os_release(host: os_utils.OsRelease, chroot: os_utils.OsRelease) -> None:
     """Compare OsRelease objects from host and chroot for compatibility. See _host_compatible_chroot."""
     if (host_val := host.id()) != (chroot_val := chroot.id()):
-        errors.IncompatibleChrootError("id", host_val, chroot_val)
+        raise errors.IncompatibleChrootError("id", host_val, chroot_val)
 
     if (host_val := host.version_id()) != (chroot_val := chroot.version_id()):
-        errors.IncompatibleChrootError("version_id", host_val, chroot_val)
+        raise errors.IncompatibleChrootError("version_id", host_val, chroot_val)
 
 
-def _host_compatible_chroot(path: Path) -> bool:
-    """Raise exception if host and chroot are not the same distribution and release"""
+def _host_compatible_chroot(path: Path) -> None:
+    """Raise exception if host and chroot are not the same distribution and release."""
     # Note: /etc/os-release is symlinked to /usr/lib/os-release
     # This could cause an issue if /etc/os-release is removed at any point.
     host_os_release = os_utils.OsRelease()
@@ -111,7 +111,7 @@ def _host_compatible_chroot(path: Path) -> bool:
     _compare_os_release(host_os_release, chroot_os_release)
 
 
-def _setup_chroot(path: Path, use_host_sources: bool) -> None:
+def _setup_chroot(path: Path, use_host_sources: bool) -> None:  # noqa: FBT001
     """Prepare the chroot environment before executing the target function."""
     logger.debug("setup chroot: %r", path)
     if sys.platform == "linux":
@@ -126,7 +126,7 @@ def _setup_chroot(path: Path, use_host_sources: bool) -> None:
     logger.debug("chroot setup complete")
 
 
-def _cleanup_chroot(path: Path, use_host_sources: bool) -> None:
+def _cleanup_chroot(path: Path, use_host_sources: bool) -> None:  # noqa: FBT001
     """Clean the chroot environment after executing the target function."""
     logger.debug("cleanup chroot: %r", path)
     if sys.platform == "linux":
@@ -140,20 +140,27 @@ def _cleanup_chroot(path: Path, use_host_sources: bool) -> None:
     logger.debug("chroot cleanup complete")
 
 
+# TODO: refactor as to not call _Mount.get_abs_path repeatedly
 class _Mount:
     def __init__(
         self,
         src: str | Path,
         dst: str | Path,
-        *args,
+        *args: str,
         fstype: str | None = None,
         skip_missing: bool = True,
     ) -> None:
-        """Management class for chroot mounts."""
-
+        """Manage setup and clean up of chroot mounts.
+
+        :param src: Mount source. This can be a device or path on the file system.
+        :param dst: Point. Path to the mount point relative to the chroot mounted on.
+        :param args: Additional args to pass to mount command.
+        :param fstype: fstype arg to use when calling mount.
+        :param skip_missing: skip mounts when dst_exists returns False.
+        """
         self.src = Path(src)
         self.dst = Path(dst)
-        self.args = list(args)
+        self.args = [*args]
         self.skip_missing = skip_missing
 
         if fstype is not None:
@@ -170,21 +177,26 @@ def _umount(self, chroot: Path, *args: str) -> None:
         os_utils.umount(str(abs_dst), *args)
 
     def get_abs_path(self, path: Path, chroot_path: Path) -> Path:
-        """Make `chroot_path` relative to host `path`"""
+        """Make `chroot_path` relative to host `path`."""
         return path / str(chroot_path).lstrip("/")
 
-    def dst_exists(self, chroot: Path):
+    def dst_exists(self, chroot: Path) -> bool:
+        """Return True if `self.dst` exists within `chroot`."""
         abs_dst = self.get_abs_path(chroot, self.dst)
         return abs_dst.is_symlink() or abs_dst.exists()
 
-    def remove_dst(self, chroot: Path):
-        pass
+    def remove_dst(self, chroot: Path) -> None:
+        """Remove `self.dst` if present to prepare mountpoint `self.dst`."""
+        # This is not required for the base class
+        # TODO: consider using abc
 
-    def create_dst(self, chroot: Path):
+    def create_dst(self, chroot: Path) -> None:
+        """Create mountpoint `self.dst` later used in mount call."""
         abs_dst = self.get_abs_path(chroot, self.dst)
         abs_dst.parent.mkdir(parents=True, exist_ok=True)
 
     def mount_to(self, chroot: Path, *args: str) -> None:
+        """Mount `self.src` to `self.dst` within chroot."""
         logger.debug(f"Mounting {self.dst}")
         if self.dst_exists(chroot):
             self.remove_dst(chroot)
@@ -198,6 +210,7 @@ def mount_to(self, chroot: Path, *args: str) -> None:
         self._mount(self.src, chroot, *self.args, *args)
 
     def unmount_from(self, chroot: Path, *args: str) -> None:
+        """Unmount `self.dst` within chroot."""
         logger.debug(f"Mounting {self.dst}")
         if self.skip_missing and not self.dst_exists(chroot):
             abs_dst = self.get_abs_path(chroot, self.dst)
@@ -217,11 +230,20 @@ def __init__(
         *args: str,
         skip_missing: bool = True,
     ) -> None:
+        """Manage setup and clean up of `--bind` mount chroot mounts.
+
+        This subclass of _Mount contains extra support for creating mount points for
+        individual files.
+        :param src: Mount source. This can be a device or path on the file system.
+        :param dst: Point. Path to the mount point relative to the chroot mounted on.
+        :param args: Additional args to pass to mount command.
+        :param skip_missing: skip mounts when dst_exists returns False.
+        """
         super().__init__(
             src, dst, f"--{self.bind_type}", *args, skip_missing=skip_missing
         )
 
-    def dst_exists(self, chroot: Path):
+    def dst_exists(self, chroot: Path) -> bool:
         abs_dst = self.get_abs_path(chroot, self.dst)
 
         if self.src.is_file():
@@ -229,7 +251,7 @@ def dst_exists(self, chroot: Path):
 
         return abs_dst.is_symlink() or abs_dst.exists()
 
-    def create_dst(self, chroot: Path):
+    def create_dst(self, chroot: Path) -> None:
         abs_dst = self.get_abs_path(chroot, self.dst)
 
         if self.src.is_dir():
@@ -237,7 +259,7 @@ def create_dst(self, chroot: Path):
         elif self.src.is_file():
             abs_dst.touch()
 
-    def remove_dst(self, chroot: Path):
+    def remove_dst(self, chroot: Path) -> None:
         abs_dst = self.get_abs_path(chroot, self.dst)
 
         if abs_dst.is_symlink() or abs_dst.is_file():
@@ -253,15 +275,26 @@ def _mount(self, src: Path, chroot: Path, *args: str) -> None:
 class _RBindMount(_BindMount):
     bind_type = "rbind"
 
-    def _umount(self, chroot: Path, *args) -> None:
+    def _umount(self, chroot: Path, *args: str) -> None:
         super()._umount(chroot, "--recursive", "--lazy", *args)
 
 
 class _TempFSClone(_Mount):
-    def __init__(self, src: str, dst: str, *args) -> None:
-        super().__init__(src, dst, *args, fstype="tmpfs", skip_missing=False)
+    def __init__(
+        self, src: str, dst: str, *args: str, skip_missing: bool = True
+    ) -> None:
+        """Manage setup and clean up of `--bind` mount chroot mounts.
 
-    def _mount(self, src: Path, chroot: Path, *args) -> None:
+        This subclass of _Mount contains extra support for creating mount points for
+        individual files.
+        :param src: Mount source. This can be a device or path on the file system.
+        :param dst: Point. Path to the mount point relative to the chroot mounted on.
+        :param args: Additional args to pass to mount command.
+        :param skip_missing: skip mounts when dst_exists returns False.
+        """
+        super().__init__(src, dst, *args, fstype="tmpfs", skip_missing=skip_missing)
+
+    def _mount(self, src: Path, chroot: Path, *args: str) -> None:
         if src.is_file():
             raise NotADirectoryError(f"Path is a file: {src}")
         if not src.exists():
@@ -293,7 +326,7 @@ def _mount(self, src: Path, chroot: Path, *args) -> None:
 # Mounts required to import host's Ubuntu Pro apt configuration to chroot
 # TODO: parameterize this per linux distribution / package manager
 _ubuntu_apt_mounts = [
-    _TempFSClone("/etc/apt", "/etc/apt"),
+    _TempFSClone("/etc/apt", "/etc/apt", skip_missing=False),
     _BindMount(
         "/usr/share/ca-certificates/",
         "/usr/share/ca-certificates/",
@@ -308,13 +341,11 @@ def _mount(self, src: Path, chroot: Path, *args) -> None:
 
 def _setup_chroot_mounts(path: Path, mounts: list[_Mount]) -> None:
     """Linux-specific chroot environment preparation."""
-
     for entry in mounts:
         entry.mount_to(path)
 
 
 def _cleanup_chroot_mounts(path: Path, mounts: list[_Mount]) -> None:
     """Linux-specific chroot environment cleanup."""
-
     for entry in reversed(mounts):
         entry.unmount_from(path)