use std::fs::File; use std::io::{Cursor, Read, Write}; use std::os::unix::ffi::OsStringExt; use std::os::unix::fs::{MetadataExt, PermissionsExt}; use std::path::{Path, PathBuf}; use std::sync::{Arc, Barrier}; use std::thread; use flate2::Compression; use flate2::write::GzEncoder; use lazaret_core::identity::{ DeviceId, HostPath, IdMapDigest, InodeId, PolicyId, PrepareOperationId, PreparedArtifactId, SelectedManifestDigest, SourceIdentity, }; use lazaret_core::records::{PrepareOperation, PrepareStageState}; use lazaret_prepared_lower::{ IdMap, IdMapRange, PrepareRecovery, PrepareRequest, PreparedLowerStore, ReclaimOutcome, }; use lazaret_storage_catalog::{Catalog, CatalogError}; use serde_json::json; use sha2::{Digest, Sha256}; use tar::{Builder, EntryType, Header}; use tempfile::TempDir; const OCI_TAR: &str = "application/vnd.oci.image.layer.v1.tar"; const OCI_GZIP: &str = "application/vnd.oci.image.layer.v1.tar+gzip"; const OCI_ZSTD: &str = "application/vnd.oci.image.layer.v1.tar+zstd"; struct Fixture { directory: TempDir, root: PathBuf, catalog: Catalog, store: PreparedLowerStore, } impl Fixture { fn new() -> Self { let directory = TempDir::new().unwrap(); let root = directory.path().join("storage"); std::fs::create_dir(&root).unwrap(); let catalog = Catalog::new(root.join("catalog.redb"), root.join("storage-catalog.lock")); catalog.initialize().unwrap(); let store = PreparedLowerStore::initialize(&root, catalog.clone()).unwrap(); Self { directory, root, catalog, store, } } fn request(&self, operation: u8, layers: &[(Vec, &str)]) -> PrepareRequest { let (layout_path, manifest) = layout(self.directory.path(), operation, layers); PrepareRequest { operation_id: PrepareOperationId::new([operation; 32]), layout_path, source_identity: b"oci-v1\0registry.invalid\0fixture\0sha256:source".to_vec(), selected_manifest_digest: manifest, id_map: current_id_map(), } } fn artifact_root(&self, artifact: &lazaret_core::records::PreparedArtifact) -> PathBuf { self.root.join(PathBuf::from(std::ffi::OsString::from_vec( artifact.path.as_bytes().to_vec(), ))) } } #[derive(Clone)] struct Entry<'a> { path: &'a str, kind: EntryType, data: &'a [u8], target: Option<&'a str>, mode: u32, } fn layer(entries: &[Entry<'_>]) -> Vec { let uid = rustix::process::geteuid().as_raw(); let gid = rustix::process::getegid().as_raw(); let mut bytes = Vec::new(); { let mut builder = Builder::new(&mut bytes); for entry in entries { let mut header = Header::new_gnu(); header.set_path(entry.path).unwrap(); header.set_entry_type(entry.kind); header.set_size(entry.data.len() as u64); header.set_uid(u64::from(uid)); header.set_gid(u64::from(gid)); header.set_mode(entry.mode); header.set_mtime(1_600_000_123); if let Some(target) = entry.target { header.set_link_name(target).unwrap(); } header.set_cksum(); builder.append(&header, Cursor::new(entry.data)).unwrap(); } builder.finish().unwrap(); } bytes } fn xattr_layer() -> Vec { let mut bytes = Vec::new(); { let mut builder = Builder::new(&mut bytes); builder .append_pax_extensions([("SCHILY.xattr.user.fixture", b"value".as_slice())]) .unwrap(); let uid = rustix::process::geteuid().as_raw(); let gid = rustix::process::getegid().as_raw(); let mut header = Header::new_gnu(); header.set_path("payload").unwrap(); header.set_entry_type(EntryType::Regular); header.set_size(1); header.set_uid(u64::from(uid)); header.set_gid(u64::from(gid)); header.set_mode(0o644); header.set_cksum(); builder.append(&header, Cursor::new(b"x")).unwrap(); builder.finish().unwrap(); } bytes } fn compress(bytes: &[u8], media_type: &str) -> Vec { match media_type { OCI_TAR => bytes.to_vec(), OCI_GZIP => { let mut encoder = GzEncoder::new(Vec::new(), Compression::default()); encoder.write_all(bytes).unwrap(); encoder.finish().unwrap() } OCI_ZSTD => zstd::encode_all(bytes, 1).unwrap(), _ => unreachable!(), } } fn layout(root: &Path, seed: u8, layers: &[(Vec, &str)]) -> (PathBuf, [u8; 32]) { let path = root.join(format!("layout-{seed}")); let blobs = path.join("blobs/sha256"); std::fs::create_dir_all(&blobs).unwrap(); let mut diff_ids = Vec::new(); let descriptors: Vec<_> = layers .iter() .map(|(bytes, media_type)| { let uncompressed = match *media_type { OCI_GZIP => { let mut output = Vec::new(); flate2::read::GzDecoder::new(bytes.as_slice()) .read_to_end(&mut output) .unwrap(); output } OCI_ZSTD => zstd::decode_all(bytes.as_slice()).unwrap(), _ => bytes.clone(), }; diff_ids.push(format!("sha256:{}", hex(&Sha256::digest(&uncompressed)))); let digest = format!("sha256:{}", hex(&Sha256::digest(bytes))); std::fs::write(blobs.join(&digest[7..]), bytes).unwrap(); json!({"mediaType": media_type, "digest": digest, "size": bytes.len()}) }) .collect(); let config = serde_json::to_vec(&json!({ "architecture": "amd64", "os": "linux", "rootfs": {"type": "layers", "diff_ids": diff_ids} })) .unwrap(); let config_digest = format!("sha256:{}", hex(&Sha256::digest(&config))); std::fs::write(blobs.join(&config_digest[7..]), &config).unwrap(); let manifest = serde_json::to_vec(&json!({ "schemaVersion": 2, "config": {"mediaType": "application/vnd.oci.image.config.v1+json", "digest": config_digest, "size": config.len()}, "layers": descriptors })) .unwrap(); let manifest_hash: [u8; 32] = Sha256::digest(&manifest).into(); let manifest_digest = format!("sha256:{}", hex(&manifest_hash)); std::fs::write(blobs.join(&manifest_digest[7..]), &manifest).unwrap(); std::fs::write( path.join("index.json"), serde_json::to_vec(&json!({ "schemaVersion": 2, "manifests": [{"mediaType": "application/vnd.oci.image.manifest.v1+json", "digest": manifest_digest, "size": manifest.len()}] })) .unwrap(), ) .unwrap(); std::fs::write( path.join("oci-layout"), b"{\"imageLayoutVersion\":\"1.0.0\"}\n", ) .unwrap(); (path, manifest_hash) } fn current_id_map() -> IdMap { IdMap { uid: vec![IdMapRange { container_id: rustix::process::geteuid().as_raw(), namespace_id: rustix::process::geteuid().as_raw(), size: 1, }], gid: vec![IdMapRange { container_id: rustix::process::getegid().as_raw(), namespace_id: rustix::process::getegid().as_raw(), size: 1, }], } } fn lower_entries() -> Vec> { vec![ Entry { path: "gone", kind: EntryType::Regular, data: b"gone", target: None, mode: 0o644, }, Entry { path: "opaque", kind: EntryType::Directory, data: b"", target: None, mode: 0o755, }, Entry { path: "opaque/old", kind: EntryType::Regular, data: b"old", target: None, mode: 0o644, }, Entry { path: "directory-to-file", kind: EntryType::Directory, data: b"", target: None, mode: 0o755, }, Entry { path: "directory-to-file/child", kind: EntryType::Regular, data: b"child", target: None, mode: 0o644, }, Entry { path: "file-to-directory", kind: EntryType::Regular, data: b"file", target: None, mode: 0o644, }, Entry { path: "target", kind: EntryType::Regular, data: b"linked", target: None, mode: 0o2755, }, ] } fn upper_entries() -> Vec> { vec![ Entry { path: ".wh.gone", kind: EntryType::Regular, data: b"", target: None, mode: 0o000, }, Entry { path: "opaque/new", kind: EntryType::Regular, data: b"new", target: None, mode: 0o644, }, Entry { path: "opaque/.wh..wh..opq", kind: EntryType::Regular, data: b"", target: None, mode: 0o000, }, Entry { path: "directory-to-file", kind: EntryType::Regular, data: b"replacement", target: None, mode: 0o600, }, Entry { path: "file-to-directory", kind: EntryType::Directory, data: b"", target: None, mode: 0o755, }, Entry { path: "file-to-directory/new", kind: EntryType::Regular, data: b"new", target: None, mode: 0o644, }, Entry { path: "symbolic", kind: EntryType::Symlink, data: b"", target: Some("target"), mode: 0o777, }, Entry { path: "hard", kind: EntryType::Link, data: b"", target: Some("target"), mode: 0o644, }, Entry { path: "forward", kind: EntryType::Link, data: b"", target: Some("later"), mode: 0o644, }, Entry { path: "later", kind: EntryType::Regular, data: b"forward", target: None, mode: 0o4755, }, ] } #[test] fn semantic_corpus_preserves_supported_changesets_and_read_only_lower_use() { let fixture = Fixture::new(); let first = layer(&lower_entries()); let second = layer(&upper_entries()); let request = fixture.request( 1, &[ (compress(&first, OCI_GZIP), OCI_GZIP), (compress(&second, OCI_ZSTD), OCI_ZSTD), ], ); let artifact = fixture.store.prepare(&request).unwrap(); let root = fixture.artifact_root(&artifact); assert!(!root.join("gone").exists()); assert!(!root.join("opaque/old").exists()); assert_eq!(std::fs::read(root.join("opaque/new")).unwrap(), b"new"); assert_eq!( std::fs::read(root.join("directory-to-file")).unwrap(), b"replacement" ); assert_eq!( std::fs::read(root.join("file-to-directory/new")).unwrap(), b"new" ); assert_eq!( std::fs::read_link(root.join("symbolic")).unwrap(), Path::new("target") ); assert_eq!( std::fs::metadata(root.join("hard")).unwrap().ino(), std::fs::metadata(root.join("target")).unwrap().ino() ); assert_eq!(std::fs::read(root.join("forward")).unwrap(), b"forward"); assert_eq!( std::fs::metadata(root.join("later")) .unwrap() .permissions() .mode() & 0o7777, 0o4755 ); std::fs::remove_dir_all(&request.layout_path).unwrap(); let mut lower = File::open(root.join("directory-to-file")).unwrap(); let mut bytes = Vec::new(); lower.read_to_end(&mut bytes).unwrap(); assert_eq!(bytes, b"replacement"); assert!( fixture .root .join("stages") .read_dir() .unwrap() .next() .is_none() ); } #[test] fn request_manifest_identity_must_match_verified_layout_manifest() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"content", target: None, mode: 0o644, }]); let mut request = fixture.request(3, &[(bytes, OCI_TAR)]); request.selected_manifest_digest = [0xff; 32]; assert!(fixture.store.prepare(&request).is_err()); let snapshot = fixture.catalog.snapshot().unwrap(); assert!(snapshot.prepared_artifacts.is_empty()); assert!(snapshot.prepare_operations.is_empty()); } #[test] fn layer_entry_matching_internal_spool_name_is_preserved() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: ".layer-0.tar", kind: EntryType::Regular, data: b"image-owned", target: None, mode: 0o640, }]); let request = fixture.request(4, &[(bytes, OCI_TAR)]); let artifact = fixture.store.prepare(&request).unwrap(); assert_eq!( std::fs::read(fixture.artifact_root(&artifact).join(".layer-0.tar")).unwrap(), b"image-owned" ); } #[test] fn root_opaque_whiteout_removes_only_lower_layer_entries() { let fixture = Fixture::new(); let lower = layer(&[Entry { path: "old", kind: EntryType::Regular, data: b"old", target: None, mode: 0o644, }]); let upper = layer(&[ Entry { path: "new", kind: EntryType::Regular, data: b"new", target: None, mode: 0o644, }, Entry { path: ".wh..wh..opq", kind: EntryType::Regular, data: b"", target: None, mode: 0o000, }, ]); let request = fixture.request(2, &[(lower, OCI_TAR), (upper, OCI_TAR)]); let artifact = fixture.store.prepare(&request).unwrap(); let root = fixture.artifact_root(&artifact); assert!(!root.join("old").exists()); assert_eq!(std::fs::read(root.join("new")).unwrap(), b"new"); } #[test] fn hostile_and_corrupt_layers_refuse_without_publication() { let cases = [ layer(&[Entry { path: "pipe", kind: EntryType::Fifo, data: b"", target: None, mode: 0o644, }]), layer(&[Entry { path: "escape", kind: EntryType::Link, data: b"", target: Some("../outside"), mode: 0o644, }]), xattr_layer(), ]; for (index, bytes) in cases.into_iter().enumerate() { let fixture = Fixture::new(); let request = fixture.request(u8::try_from(index).unwrap() + 10, &[(bytes, OCI_TAR)]); assert!(fixture.store.prepare(&request).is_err()); let snapshot = fixture.catalog.snapshot().unwrap(); assert!(snapshot.prepared_artifacts.is_empty()); assert!(snapshot.prepare_operations.is_empty()); assert!( fixture .root .join("stages") .read_dir() .unwrap() .next() .is_none() ); } let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "file", kind: EntryType::Regular, data: b"content", target: None, mode: 0o644, }]); let request = fixture.request(20, &[(bytes, OCI_TAR)]); let blob = request .layout_path .join("blobs/sha256") .read_dir() .unwrap() .map(|entry| entry.unwrap().path()) .find(|path| std::fs::metadata(path).unwrap().len() > 1024) .unwrap(); OpenOptionsExt::append_byte(&blob); assert!(fixture.store.prepare(&request).is_err()); assert!( fixture .catalog .snapshot() .unwrap() .prepared_artifacts .is_empty() ); } struct OpenOptionsExt; impl OpenOptionsExt { fn append_byte(path: &Path) { let mut file = std::fs::OpenOptions::new().append(true).open(path).unwrap(); file.write_all(b"x").unwrap(); } } #[test] fn high_expansion_layer_is_disk_spooled_and_published() { let fixture = Fixture::new(); let payload = vec![0u8; 32 * 1024 * 1024]; let tar = layer(&[Entry { path: "expanded", kind: EntryType::Regular, data: &payload, target: None, mode: 0o600, }]); let compressed = compress(&tar, OCI_GZIP); assert!(compressed.len() * 100 < payload.len()); let request = fixture.request(30, &[(compressed, OCI_GZIP)]); let artifact = fixture.store.prepare(&request).unwrap(); assert_eq!( std::fs::metadata(fixture.artifact_root(&artifact).join("expanded")) .unwrap() .len(), payload.len() as u64 ); assert!( fixture .root .join("stages") .read_dir() .unwrap() .next() .is_none() ); } #[test] fn concurrent_publication_reuses_one_immutable_artifact() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"same", target: None, mode: 0o644, }]); let first = fixture.request(40, &[(bytes.clone(), OCI_TAR)]); let mut second = PrepareRequest { operation_id: PrepareOperationId::new([41; 32]), ..first.clone() }; second.layout_path = first.layout_path.clone(); let store = Arc::new(fixture.store.clone()); let barrier = Arc::new(Barrier::new(3)); let first_handle = { let store = Arc::clone(&store); let barrier = Arc::clone(&barrier); thread::spawn(move || { barrier.wait(); store.prepare(&first) }) }; let second_handle = { let store = Arc::clone(&store); let barrier = Arc::clone(&barrier); thread::spawn(move || { barrier.wait(); store.prepare(&second) }) }; barrier.wait(); let first = first_handle.join().unwrap().unwrap(); let second = second_handle.join().unwrap().unwrap(); assert_eq!(first, second); let snapshot = fixture.catalog.snapshot().unwrap(); assert_eq!(snapshot.prepared_artifacts, vec![first]); assert!(snapshot.prepare_operations.is_empty()); } #[test] fn publication_during_reclaim_preserves_replacement_and_deletes_old_inode() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"same", target: None, mode: 0o644, }]); let first_request = fixture.request(50, &[(bytes, OCI_TAR)]); let first = fixture.store.prepare(&first_request).unwrap(); let reclaiming = fixture.catalog.begin_reclaim(&first).unwrap(); let replacement_request = PrepareRequest { operation_id: PrepareOperationId::new([51; 32]), ..first_request }; let barrier = Arc::new(Barrier::new(3)); let store = Arc::new(fixture.store.clone()); let reclaim_handle = { let barrier = Arc::clone(&barrier); let store = Arc::clone(&store); thread::spawn(move || { barrier.wait(); store.resume_reclaim(&reclaiming) }) }; let publish_handle = { let barrier = Arc::clone(&barrier); let store = Arc::clone(&store); thread::spawn(move || { barrier.wait(); store.prepare(&replacement_request) }) }; barrier.wait(); let reclaim = reclaim_handle.join().unwrap().unwrap(); let replacement = publish_handle.join().unwrap().unwrap(); assert!(matches!( reclaim, ReclaimOutcome::Removed | ReclaimOutcome::ReplacementPreserved )); assert_ne!(replacement.inode, first.inode); assert_eq!( fixture.catalog.snapshot().unwrap().prepared_artifacts, vec![replacement] ); assert!( fixture .root .join("quarantine") .read_dir() .unwrap() .next() .is_none() ); } #[test] fn replacement_winning_before_reclaimer_resumes_still_cleans_old_quarantine() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"same", target: None, mode: 0o644, }]); let first_request = fixture.request(52, &[(bytes, OCI_TAR)]); let first = fixture.store.prepare(&first_request).unwrap(); let reclaiming = fixture.catalog.begin_reclaim(&first).unwrap(); let replacement_request = PrepareRequest { operation_id: PrepareOperationId::new([53; 32]), ..first_request }; let replacement = fixture.store.prepare(&replacement_request).unwrap(); assert_ne!(replacement.inode, first.inode); assert_eq!( fixture.store.resume_reclaim(&reclaiming).unwrap(), ReclaimOutcome::ReplacementPreserved ); assert_eq!( fixture.catalog.snapshot().unwrap().prepared_artifacts, vec![replacement.clone()] ); assert!(fixture.artifact_root(&replacement).exists()); assert!( fixture .root .join("quarantine") .read_dir() .unwrap() .next() .is_none() ); } #[test] fn reclaim_refuses_preexisting_foreign_quarantine_path() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"same", target: None, mode: 0o644, }]); let artifact = fixture .store .prepare(&fixture.request(54, &[(bytes, OCI_TAR)])) .unwrap(); let foreign = fixture.root.join("quarantine").join(format!( "{}-{}-{}", hex(artifact.prepared_artifact_id.as_bytes()), artifact.device.get(), artifact.inode.get() )); std::fs::create_dir(&foreign).unwrap(); assert!(fixture.store.reclaim(&artifact).is_err()); assert!(foreign.is_dir()); assert!(fixture.artifact_root(&artifact).is_dir()); } #[test] fn recovery_deletes_only_exact_owned_stage_and_preserves_foreign_stage() { let fixture = Fixture::new(); let operation_id = PrepareOperationId::new([60; 32]); let stage = Path::new("stages").join(hex(operation_id.as_bytes())); let intent = PrepareOperation { operation_id, expected_artifact_id: PreparedArtifactId::new([61; 32]), source_identity: SourceIdentity::new(b"source".to_vec()), selected_manifest_digest: SelectedManifestDigest::new([62; 32]), policy_id: PolicyId::new([63; 32]), id_map_digest: IdMapDigest::new([64; 32]), stage_path: HostPath::new(stage.as_os_str().as_encoded_bytes().to_vec()), stage_state: PrepareStageState::Intent, device: None, inode: None, }; fixture.catalog.record_prepare(&intent).unwrap(); std::fs::create_dir(fixture.root.join(&stage)).unwrap(); std::fs::write(fixture.root.join(&stage).join("foreign"), b"keep").unwrap(); assert!(fixture.store.recover_prepare(&intent).is_err()); assert!(fixture.root.join(&stage).join("foreign").exists()); std::fs::remove_dir_all(fixture.root.join(&stage)).unwrap(); assert_eq!( fixture.store.recover_prepare(&intent).unwrap(), PrepareRecovery::ClearedUnmaterializedIntent ); let owned_id = PrepareOperationId::new([65; 32]); let owned_stage = Path::new("stages").join(hex(owned_id.as_bytes())); let owned_intent = PrepareOperation { operation_id: owned_id, stage_path: HostPath::new(owned_stage.as_os_str().as_encoded_bytes().to_vec()), ..intent }; fixture.catalog.record_prepare(&owned_intent).unwrap(); std::fs::create_dir(fixture.root.join(&owned_stage)).unwrap(); std::fs::write(fixture.root.join(&owned_stage).join("partial"), b"partial").unwrap(); let metadata = std::fs::metadata(fixture.root.join(&owned_stage)).unwrap(); let owned = PrepareOperation { stage_state: PrepareStageState::Owned, device: Some(DeviceId::new(metadata.dev())), inode: Some(InodeId::new(metadata.ino())), ..owned_intent.clone() }; fixture .catalog .bind_prepare_stage(&owned_intent, &owned) .unwrap(); assert_eq!( fixture.store.recover_prepare(&owned).unwrap(), PrepareRecovery::RemovedOwnedStage ); assert!(!fixture.root.join(&owned_stage).exists()); } #[test] fn reclaim_refuses_forged_identity_and_leases() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"same", target: None, mode: 0o644, }]); let artifact = fixture .store .prepare(&fixture.request(70, &[(bytes, OCI_TAR)])) .unwrap(); let lease = lazaret_core::records::ArtifactLease { prepared_artifact_id: artifact.prepared_artifact_id, domain_id: lazaret_core::identity::DomainId::new([71; 32]), generation_id: lazaret_core::identity::GenerationId::new(1), }; fixture.catalog.acquire_lease(&lease).unwrap(); assert!(matches!( fixture.store.reclaim(&artifact), Err(error) if matches!(error.downcast_ref::(), Some(CatalogError::LeasesPresent)) )); fixture.catalog.release_lease(&lease).unwrap(); let mut forged = artifact.clone(); forged.inode = InodeId::new(artifact.inode.get() + 1); assert!(fixture.store.reclaim(&forged).is_err()); assert!(fixture.artifact_root(&artifact).exists()); } fn hex(bytes: &[u8]) -> String { use std::fmt::Write as _; bytes.iter().fold( String::with_capacity(bytes.len() * 2), |mut output, byte| { write!(output, "{byte:02x}").expect("writing to String cannot fail"); output }, ) }