wip: add test for concurrent image layer threading

Author: manuelh-dev

image-rs/tests/pull_luks_integrity_linux.rs

@@ -0,0 +1,228 @@
+// Copyright (c) 2026 The image-rs Authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//! Reproduces the trusted image store scenario from
+//! <https://github.com/kata-containers/kata-containers/issues/12721>:
+//! concurrent layer unpack on ext4 over LUKS2 + dm-integrity.
+//!
+//! Requires root, `cryptsetup`, `mkfs.ext4`, and loop device support.
+//! Run explicitly:
+//! `cargo test -p image-rs pull_image_on_luks2_integrity_parallel_layers -- --ignored --nocapture`
+
+#[cfg(not(target_os = "linux"))]
+#[test]
+fn pull_luks_integrity_linux_not_run_on_non_linux() {}
+
+#[cfg(target_os = "linux")]
+use std::io::Write;
+#[cfg(target_os = "linux")]
+use std::path::Path;
+#[cfg(target_os = "linux")]
+use std::process::Command;
+#[cfg(target_os = "linux")]
+use std::time::{SystemTime, UNIX_EPOCH};
+
+#[cfg(target_os = "linux")]
+use loopdev::LoopControl;
+#[cfg(target_os = "linux")]
+use nix::mount::{mount, MsFlags};
+#[cfg(target_os = "linux")]
+use tempfile::NamedTempFile;
+
+#[cfg(target_os = "linux")]
+const BACKING_SIZE: u64 = 512 * 1024 * 1024;
+#[cfg(target_os = "linux")]
+const PASSPHRASE: &[u8] = b"image-rs-luks-integrity-test";
+/// Image with multiple layers so `buffer_unordered` can run concurrently.
+#[cfg(target_os = "linux")]
+const TEST_IMAGE: &str = "ghcr.io/confidential-containers/test-container-image-rs:busybox-gzip";
+
+#[cfg(target_os = "linux")]
+fn run_cryptsetup(args: &[&str]) -> std::io::Result<()> {
+    let status = Command::new("cryptsetup").args(args).status()?;
+    if !status.success() {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::Other,
+            format!("cryptsetup {:?} failed: {status}", args.first()),
+        ));
+    }
+    Ok(())
+}
+
+#[cfg(target_os = "linux")]
+fn which_cryptsetup() -> bool {
+    Command::new("which")
+        .arg("cryptsetup")
+        .output()
+        .map(|o| o.status.success())
+        .unwrap_or(false)
+}
+
+#[cfg(target_os = "linux")]
+struct LuksIntegrityMount {
+    mount_point: tempfile::TempDir,
+    mapper_name: String,
+    loop_dev: loopdev::LoopDevice,
+}
+
+#[cfg(target_os = "linux")]
+impl LuksIntegrityMount {
+    fn new() -> Result<Self, String> {
+        if !nix::unistd::Uid::effective().is_root() {
+            return Err("must run as root".into());
+        }
+        if !which_cryptsetup() {
+            return Err("cryptsetup not found in PATH".into());
+        }
+
+        let backing = NamedTempFile::new().map_err(|e| e.to_string())?;
+        backing
+            .as_file()
+            .set_len(BACKING_SIZE)
+            .map_err(|e| e.to_string())?;
+
+        let loop_control = LoopControl::open().map_err(|e| e.to_string())?;
+        let loop_dev = loop_control.next_free().map_err(|e| e.to_string())?;
+        loop_dev
+            .with()
+            .autoclear(true)
+            .attach(
+                backing
+                    .path()
+                    .to_str()
+                    .ok_or_else(|| "backing path".to_string())?,
+            )
+            .map_err(|e| e.to_string())?;
+
+        let loop_path = loop_dev
+            .path()
+            .ok_or_else(|| "loop device path".to_string())?;
+        let loop_path_str = loop_path
+            .to_str()
+            .ok_or_else(|| "loop path utf-8".to_string())?;
+
+        let nanos = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map(|d| d.as_nanos())
+            .unwrap_or(0);
+        let mapper_name = format!("img_rs_luks_{}_{}", std::process::id(), nanos);
+
+        let mut keyfile = NamedTempFile::new().map_err(|e| e.to_string())?;
+        keyfile
+            .write_all(PASSPHRASE)
+            .map_err(|e| e.to_string())?;
+        keyfile.flush().map_err(|e| e.to_string())?;
+        let key_path = keyfile
+            .path()
+            .to_str()
+            .ok_or_else(|| "key path".to_string())?;
+
+        // Align with confidential-data-hub LUKS2 + integrity (sector 4096, hmac-sha256).
+        run_cryptsetup(&[
+            "luksFormat",
+            "--type",
+            "luks2",
+            "--integrity",
+            "hmac-sha256",
+            "--sector-size",
+            "4096",
+            "--batch-mode",
+            "--pbkdf",
+            "pbkdf2",
+            loop_path_str,
+            "--key-file",
+            key_path,
+        ])
+        .map_err(|e| e.to_string())?;
+
+        // Match CDH `CryptActivate::NO_JOURNAL` for dm-integrity.
+        run_cryptsetup(&[
+            "open",
+            "--type",
+            "luks2",
+            "--integrity-no-journal",
+            loop_path_str,
+            &mapper_name,
+            "--key-file",
+            key_path,
+        ])
+        .map_err(|e| e.to_string())?;
+
+        let mapper_path = format!("/dev/mapper/{mapper_name}");
+        let mkfs = Command::new("mkfs.ext4")
+            .args(["-F", &mapper_path])
+            .output()
+            .map_err(|e| e.to_string())?;
+        if !mkfs.status.success() {
+            return Err(format!(
+                "mkfs.ext4: {}",
+                String::from_utf8_lossy(&mkfs.stderr)
+            ));
+        }
+
+        let mount_point = tempfile::tempdir().map_err(|e| e.to_string())?;
+        nix::mount::mount(
+            Some(mapper_path.as_str()),
+            mount_point.path(),
+            Some("ext4"),
+            MsFlags::empty(),
+            None::<&str>,
+        )
+        .map_err(|e| format!("mount ext4: {e}"))?;
+
+        Ok(Self {
+            mount_point,
+            mapper_name,
+            loop_dev,
+        })
+    }
+
+    fn work_dir(&self) -> &Path {
+        self.mount_point.path()
+    }
+}
+
+#[cfg(target_os = "linux")]
+impl Drop for LuksIntegrityMount {
+    fn drop(&mut self) {
+        let _ = nix::mount::umount(self.mount_point.path());
+        let _ = Command::new("cryptsetup")
+            .args(["close", &self.mapper_name])
+            .status();
+        let _ = self.loop_dev.detach();
+    }
+}
+
+#[cfg(target_os = "linux")]
+#[tokio::test]
+#[serial_test::serial]
+#[ignore = "requires root, cryptsetup, mkfs.ext4, loop; run with: cargo test -p image-rs pull_image_on_luks2_integrity_parallel_layers -- --ignored"]
+async fn pull_image_on_luks2_integrity_parallel_layers() {
+    let store = LuksIntegrityMount::new().expect("setup LUKS+integrity mount");
+
+    let work_dir = store.work_dir().to_path_buf();
+
+    let mut image_client = image_rs::builder::ClientBuilder::default()
+        .work_dir(work_dir.clone())
+        .max_concurrent_layer_downloads_per_image(3)
+        .build()
+        .await
+        .expect("build ImageClient");
+
+    let bundle_dir = tempfile::tempdir().unwrap();
+    image_client
+        .pull_image(TEST_IMAGE, bundle_dir.path(), &None, &None)
+        .await
+        .unwrap_or_else(|e| panic!("pull_image on LUKS+integrity failed: {e:#}"));
+
+    let busybox = bundle_dir.path().join("rootfs/bin/busybox");
+    assert!(
+        busybox.exists(),
+        "expected busybox at {}",
+        busybox.display()
+    );
+
+    let mounted_rootfs = bundle_dir.path().join("rootfs");
+    let _ = nix::mount::umount(&mounted_rootfs);
+}