jailer mount propagation test

this test creates two mounts for a rootfs and guest kernel in
directories that are mounted in a location that the jailer
requires to boot successfully
Signed-off-by: Anthony Corletti <[email protected]>

Author: anthonycorletti

tests/framework/microvm.py

@@ -18,6 +18,7 @@
 import select
 import shutil
 import signal
+import subprocess
 import time
 import uuid
 from collections import namedtuple
@@ -1115,13 +1116,36 @@ def build_from_snapshot(self, snapshot: Snapshot):
         vm.restore_from_snapshot(snapshot, resume=True)
         return vm
 
+    def unmount(self, path: str) -> None:
+        """Unmounts a path with `umount` in a subprocess"""
+        try:
+            subprocess.run(["umount", path], check=True)
+        except subprocess.CalledProcessError:
+            print(f"Failed to unmount {path}")
+
+    def get_mounts_at_path(self, path: str) -> list:
+        """Get all mounts for a given path. Returns a list of mount points."""
+        try:
+            with open("/proc/mounts", "r", encoding="utf-8") as f:
+                return [
+                    line.split()[1]
+                    for line in f
+                    if line.split()[1].startswith(os.path.abspath(path))
+                ]
+        except FileNotFoundError:
+            return False  # /proc/mounts may not exist on some systems
+
     def kill(self):
         """Clean up all built VMs"""
         for vm in self.vms:
             vm.kill()
             vm.jailer.cleanup()
             chroot_base_with_id = vm.jailer.chroot_base_with_id()
             if len(vm.jailer.jailer_id) > 0 and chroot_base_with_id.exists():
+                mounts = self.get_mounts_at_path(chroot_base_with_id)
+                if mounts:
+                    for mounted_path in mounts:
+                        self.unmount(mounted_path)
                 shutil.rmtree(chroot_base_with_id)
             vm.netns.cleanup()
 

tests/integration_tests/security/test_jail.py

@@ -664,3 +664,68 @@ def test_cgroupsv2_written_only_once(uvm_plain, cgroups_info):
     assert len(write_lines) == 1
     assert len(mkdir_lines) != len(cgroups), "mkdir equal to number of cgroups"
     assert len(mkdir_lines) == 1
+
+
+def test_jail_mount(uvm_plain):
+    """
+    Test that the jailer mounts are propagated to the root mount namespace.
+    """
+    # setup the microvm
+    test_microvm = uvm_plain
+
+    chroot_base = test_microvm.jailer.chroot_base
+    # make a directory to hold the original content
+    original_content_dir = chroot_base / "original_content"
+    original_content_dir.mkdir(parents=True, exist_ok=True)
+
+    # make a directory to hold the jailed content
+    jailed_content_dir = Path(test_microvm.jailer.chroot_path())
+    jailed_content_dir.mkdir(parents=True, exist_ok=True)
+
+    # assert that the directory was created
+    assert original_content_dir.exists()
+    assert jailed_content_dir.exists()
+
+    # create the files that will be mounted
+    test_data = original_content_dir / "test_data"
+    test_data.touch()
+    assert test_data.exists()
+    test_data.write_text("test_data")
+    assert test_data.read_text() == "test_data"
+
+    jailed_test_data = jailed_content_dir / "test_data"
+    jailed_test_data.touch()
+    assert jailed_test_data.exists()
+    assert jailed_test_data.read_text() == ""
+
+    # mount the data
+    subprocess.run(["mount", "--bind", test_data, jailed_test_data], check=True)
+
+    # spawn the microvm
+    test_microvm.spawn()
+    test_microvm.basic_config()
+
+    # set params for the microvm
+    test_microvm.jailer.gid = 0
+    test_microvm.jailer.uid = 0
+    test_microvm.jailer.daemonize = True
+    test_microvm.extra_args = {"seccomp-level": 0}
+    test_microvm.add_net_iface()
+    test_microvm.start()
+
+    # mock jailer
+    for cmd in [
+        "unshare --mount --propagation unchanged",
+        "mount --make-rslave /",
+        f"mount --rbind {jailed_content_dir} {jailed_content_dir}",
+    ]:
+        subprocess.run(cmd.split(), check=True, capture_output=True)
+
+    # check that the file output is there
+    output = subprocess.run(
+        f"cat {jailed_content_dir}/test_data",
+        shell=True,
+        check=True,
+        capture_output=True,
+    )
+    assert output.stdout.decode() == "test_data"