ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Expose typed Prepared Lower failures

Browse source
This commit is contained in:
ms2k 2026-07-19 03:18:50 +09:00
commit 30cd36dcdd
9 changed files with 694 additions and 249 deletions
Download patch file Download diff file Expand all files Collapse all files

42
crates/lazaret-host-coordinator/src/runtime_selector.rs
View file

@ -17,7 +17,7 @@ use lazaret_linux::filesystem::namespace_identity;
use lazaret_linux::platform::{PlatformObservation, check_platform};
use lazaret_prepared_lower::{
IdMap, IdMapRange, OciRegistryTransport, Platform, PreparedDestination,
PreparedDestinationKind, PreparedLowerStore, RegistryCredentials,
PreparedDestinationKind, PreparedLowerError, PreparedLowerStore, RegistryCredentials,
};
use lazaret_runtime_supervisor::network::{DnsProbeTarget, NetworkProbeRequest, TcpProbeTarget};
use lazaret_runtime_supervisor::sandbox::{
@ -36,18 +36,37 @@ use crate::sandbox_init::PRIVATE_SANDBOX_INIT_ROLE;
const NETWORK_NAMESPACE_PATH: &str = "/proc/self/ns/net";
#[derive(Debug)]
pub struct RuntimeSelectionError(String);
pub enum RuntimeSelectionError {
PreparedLower(PreparedLowerError),
Other(String),
}
impl fmt::Display for RuntimeSelectionError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.0)
match self {
Self::PreparedLower(error) => error.fmt(formatter),
Self::Other(message) => formatter.write_str(message),
}
}
}
impl std::error::Error for RuntimeSelectionError {}
impl std::error::Error for RuntimeSelectionError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::PreparedLower(error) => Some(error),
Self::Other(_) => None,
}
}
}
impl From<PreparedLowerError> for RuntimeSelectionError {
fn from(error: PreparedLowerError) -> Self {
Self::PreparedLower(error)
}
}
fn failure(error: impl fmt::Display) -> RuntimeSelectionError {
RuntimeSelectionError(error.to_string())
RuntimeSelectionError::Other(error.to_string())
}
pub struct InstalledProductPaths<'a> {
@ -149,8 +168,7 @@ async fn select_runtime_for_platform(
Err(error) => return Err(failure(error)),
}
let prepared =
PreparedLowerStore::under_storage_root(configuration.storage_root(), catalog.clone())
.map_err(failure)?;
PreparedLowerStore::under_storage_root(configuration.storage_root(), catalog.clone())?;
recover_before_cold_start(configuration, &store, &catalog, &prepared).map_err(failure)?;
let image = OciRegistryTransport
.pull_confined(
@ -161,7 +179,7 @@ async fn select_runtime_for_platform(
random_id()?,
)
.await
.map_err(|error| failure(format!("pulling configured anonymous OCI image: {error:#}")))?;
.map_err(RuntimeSelectionError::PreparedLower)?;
let artifact = prepared
.prepare_transported(
PrepareOperationId::new(random_id()?),
@ -169,10 +187,10 @@ async fn select_runtime_for_platform(
rootless_id_map(),
prepared_destinations(configuration),
)
.map_err(failure)?;
.map_err(RuntimeSelectionError::PreparedLower)?;
let artifact_authority = prepared
.open_artifact_authority(&artifact)
.map_err(failure)?;
.map_err(RuntimeSelectionError::PreparedLower)?;
let artifact_label = PathBuf::from(std::ffi::OsString::from_vec(
configuration
.storage_root()
@ -184,7 +202,9 @@ async fn select_runtime_for_platform(
.join(PathBuf::from(std::ffi::OsString::from_vec(
artifact.path.as_bytes().to_vec(),
)));
image.cleanup().map_err(failure)?;
image
.cleanup()
.map_err(RuntimeSelectionError::PreparedLower)?;
let next_generation = next_generation(&store)?;
let (coordinator_id, network_sidecar_id, sandbox_id) = distinct_generation_ids()?;

161
crates/lazaret-prepared-lower/src/error.rs Normal file
View file

@ -0,0 +1,161 @@
use std::fmt;
pub struct PreparedLowerSource(anyhow::Error);
impl fmt::Debug for PreparedLowerSource {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
self.0.fmt(formatter)
}
}
impl fmt::Display for PreparedLowerSource {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
if formatter.alternate() {
write!(formatter, "{:#}", self.0)
} else {
self.0.fmt(formatter)
}
}
}
impl std::error::Error for PreparedLowerSource {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
Some(self.0.as_ref())
}
}
macro_rules! prepared_lower_errors {
($($variant:ident, $constructor:ident => $label:literal),+ $(,)?) => {
#[derive(Debug)]
pub enum PreparedLowerError {
$($variant { source: PreparedLowerSource },)+
}
impl PreparedLowerError {
$(pub(crate) fn $constructor(source: impl Into<anyhow::Error>) -> Self {
Self::$variant {
source: PreparedLowerSource(source.into()),
}
})+
fn source_error(&self) -> &PreparedLowerSource {
match self {
$(Self::$variant { source } => source,)+
}
}
pub fn downcast_ref<E>(&self) -> Option<&E>
where
E: fmt::Display + fmt::Debug + Send + Sync + 'static,
{
self.source_error().0.downcast_ref()
}
}
impl fmt::Display for PreparedLowerError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
let label = match self {
$(Self::$variant { .. } => $label,)+
};
if formatter.alternate() {
write!(formatter, "{label}: {:#}", self.source_error())
} else {
write!(formatter, "{label}: {}", self.source_error())
}
}
}
};
}
prepared_lower_errors! {
Transport, transport => "OCI transport failed",
Authentication, authentication => "OCI authentication failed",
ManifestSelection, manifest_selection => "OCI manifest selection failed",
PlatformSelection, platform_selection => "OCI platform selection failed",
LayerValidation, layer_validation => "prepared layer validation failed",
Materialization, materialization => "prepared lower materialization failed",
Publication, publication => "prepared lower publication failed",
Catalog, catalog => "prepared lower catalog operation failed",
Locking, locking => "prepared lower lock acquisition failed",
Recovery, recovery => "prepared lower recovery failed",
Reclaim, reclaim => "prepared lower reclaim failed",
Confinement, confinement => "prepared lower confinement failed",
OperatingSystem, operating_system => "prepared lower operating-system operation failed",
}
impl std::error::Error for PreparedLowerError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
Some(self.source_error())
}
}
pub type PreparedLowerResult<T> = Result<T, PreparedLowerError>;
#[cfg(test)]
mod tests {
use std::error::Error as _;
use std::path::Path;
use lazaret_core::records::{PrepareOperation, PreparedArtifact};
use super::*;
use crate::{
OciRegistryTransport, Platform, PrepareRequest, PreparedLowerStore, RegistryCredentials,
TransportedImage,
};
#[test]
fn every_public_failure_category_retains_its_source() {
let errors = [
PreparedLowerError::transport(anyhow::anyhow!("source")),
PreparedLowerError::authentication(anyhow::anyhow!("source")),
PreparedLowerError::manifest_selection(anyhow::anyhow!("source")),
PreparedLowerError::platform_selection(anyhow::anyhow!("source")),
PreparedLowerError::layer_validation(anyhow::anyhow!("source")),
PreparedLowerError::materialization(anyhow::anyhow!("source")),
PreparedLowerError::publication(anyhow::anyhow!("source")),
PreparedLowerError::catalog(anyhow::anyhow!("source")),
PreparedLowerError::locking(anyhow::anyhow!("source")),
PreparedLowerError::recovery(anyhow::anyhow!("source")),
PreparedLowerError::reclaim(anyhow::anyhow!("source")),
PreparedLowerError::confinement(anyhow::anyhow!("source")),
PreparedLowerError::operating_system(anyhow::anyhow!("source")),
];
for error in errors {
assert!(error.source().is_some());
assert!(format!("{error:#}").contains("source"));
}
}
#[test]
fn exported_operations_have_typed_results() {
fn prepare(
store: &PreparedLowerStore,
request: &PrepareRequest,
) -> PreparedLowerResult<PreparedArtifact> {
store.prepare(request)
}
fn recover(
store: &PreparedLowerStore,
operation: &PrepareOperation,
) -> PreparedLowerResult<crate::PrepareRecovery> {
store.recover_prepare(operation)
}
async fn pull(
transport: &OciRegistryTransport,
reference: &str,
credentials: &RegistryCredentials,
platform: &Platform,
layout: &Path,
) -> PreparedLowerResult<TransportedImage> {
transport
.pull(reference, credentials, platform, layout)
.await
}
let _ = prepare;
let _ = recover;
let _ = pull;
}
}

2
crates/lazaret-prepared-lower/src/lib.rs
View file

@ -1,9 +1,11 @@
#![doc = "OCI transport, preparation, publication, and exact-owned reclaim for prepared lowers."]
mod error;
mod layer;
mod store;
mod transport;
pub use error::{PreparedLowerError, PreparedLowerResult, PreparedLowerSource};
pub use layer::{
IdMap, IdMapRange, PreparationLimits, PreparedDestination, PreparedDestinationKind,
};

161
crates/lazaret-prepared-lower/src/store.rs
View file

@ -28,6 +28,7 @@ use crate::layer::{
verify_layout,
};
use crate::transport::TransportedImage;
use crate::{PreparedLowerError, PreparedLowerResult};
const PHYSICAL_LOCK_WAIT: Duration = Duration::from_secs(5);
const LOCK_RETRY: Duration = Duration::from_millis(2);
@ -74,7 +75,11 @@ impl PreparedLowerStore {
/// # Errors
///
/// Returns an error if the root cannot be opened or any fixed path is not a real directory.
pub fn initialize(root: impl Into<PathBuf>, catalog: Catalog) -> Result<Self> {
pub fn initialize(root: impl Into<PathBuf>, catalog: Catalog) -> PreparedLowerResult<Self> {
Self::initialize_inner(root, catalog).map_err(PreparedLowerError::confinement)
}
fn initialize_inner(root: impl Into<PathBuf>, catalog: Catalog) -> Result<Self> {
let root = root.into();
let root = File::open(&root).context("opening Storage Root")?;
Self::initialize_retained(root, catalog)
@ -84,7 +89,15 @@ impl PreparedLowerStore {
///
/// # Errors
/// Returns an error if Storage Root authority drifts or a fixed child is not a real directory.
pub fn under_storage_root(storage_root: &ConfinedPathPlan, catalog: Catalog) -> Result<Self> {
pub fn under_storage_root(
storage_root: &ConfinedPathPlan,
catalog: Catalog,
) -> PreparedLowerResult<Self> {
Self::under_storage_root_inner(storage_root, catalog)
.map_err(PreparedLowerError::confinement)
}
fn under_storage_root_inner(storage_root: &ConfinedPathPlan, catalog: Catalog) -> Result<Self> {
let storage_root = File::from(
storage_root
.open_authority()
@ -127,7 +140,15 @@ impl PreparedLowerStore {
///
/// # Errors
/// Refuses a non-canonical artifact record, catalog drift, or physical identity drift.
pub fn open_artifact_authority(&self, artifact: &PreparedArtifact) -> Result<File> {
pub fn open_artifact_authority(
&self,
artifact: &PreparedArtifact,
) -> PreparedLowerResult<File> {
self.open_artifact_authority_inner(artifact)
.map_err(PreparedLowerError::confinement)
}
fn open_artifact_authority_inner(&self, artifact: &PreparedArtifact) -> Result<File> {
let catalog_artifact = self
.catalog_artifact(artifact.prepared_artifact_id)?
.context("prepared artifact is absent from the catalog")?;
@ -145,7 +166,7 @@ impl PreparedLowerStore {
///
/// Refuses corrupt or unsupported OCI input, forged physical ownership, catalog conflicts,
/// resource limits, and publication failures.
pub fn prepare(&self, request: &PrepareRequest) -> Result<PreparedArtifact> {
pub fn prepare(&self, request: &PrepareRequest) -> PreparedLowerResult<PreparedArtifact> {
self.prepare_from_source(request, &LayoutSource::Ambient(&request.layout_path))
}
@ -159,10 +180,21 @@ impl PreparedLowerStore {
image: &TransportedImage,
id_map: IdMap,
destinations: Vec<PreparedDestination>,
) -> Result<PreparedArtifact> {
) -> PreparedLowerResult<PreparedArtifact> {
self.prepare_transported_inner(operation_id, image, id_map, destinations)
}
fn prepare_transported_inner(
&self,
operation_id: PrepareOperationId,
image: &TransportedImage,
id_map: IdMap,
destinations: Vec<PreparedDestination>,
) -> PreparedLowerResult<PreparedArtifact> {
let layout = image
.retained_layout()
.context("transported image has no retained layout authority")?;
.context("transported image has no retained layout authority")
.map_err(PreparedLowerError::confinement)?;
let request = PrepareRequest {
operation_id,
layout_path: image.layout_path.clone(),
@ -178,32 +210,46 @@ impl PreparedLowerStore {
&self,
request: &PrepareRequest,
layout: &LayoutSource<'_>,
) -> Result<PreparedArtifact> {
) -> PreparedLowerResult<PreparedArtifact> {
if request.source_identity.is_empty() || request.source_identity.len() > 4096 {
bail!("source identity is empty or exceeds 4096 bytes");
return Err(PreparedLowerError::layer_validation(anyhow::anyhow!(
"source identity is empty or exceeds 4096 bytes"
)));
}
let verified = verify_layout(layout, self.limits)?;
let verified =
verify_layout(layout, self.limits).map_err(PreparedLowerError::layer_validation)?;
if request.selected_manifest_digest != verified.manifest_digest {
bail!("selected manifest identity does not match verified OCI layout");
return Err(PreparedLowerError::manifest_selection(anyhow::anyhow!(
"selected manifest identity does not match verified OCI layout"
)));
}
let identities = Identities::new(request)?;
if let Some(existing) = self.available_artifact(identities.artifact_id)? {
identities.require_equivalent(&existing)?;
self.require_artifact_identity(&existing)?;
let identities = Identities::new(request).map_err(PreparedLowerError::layer_validation)?;
if let Some(existing) = self
.available_artifact(identities.artifact_id)
.map_err(PreparedLowerError::catalog)?
{
identities
.require_equivalent(&existing)
.map_err(PreparedLowerError::catalog)?;
self.require_artifact_identity(&existing)
.map_err(PreparedLowerError::confinement)?;
return Ok(existing);
}
let stage_relative = stage_path(request.operation_id);
let intent = identities.intent(request.operation_id, &stage_relative);
self.catalog.record_prepare(&intent)?;
self.catalog
.record_prepare(&intent)
.map_err(PreparedLowerError::catalog)?;
let stage = match self.create_stage(&stage_relative) {
Ok(stage) => stage,
Err(error) => {
let _ = self.catalog.remove_prepare(&intent);
return Err(error);
return Err(PreparedLowerError::materialization(error));
}
};
let (device, inode) = file_identity(&stage)?;
let (device, inode) =
file_identity(&stage).map_err(PreparedLowerError::operating_system)?;
let owned = PrepareOperation {
stage_state: PrepareStageState::Owned,
device: Some(device),
@ -217,7 +263,7 @@ impl PreparedLowerStore {
{
let _ = self.catalog.remove_prepare(&intent);
}
return Err(error.into());
return Err(PreparedLowerError::catalog(error));
}
if let Err(error) = apply_layout(layout, &stage, &verified, &request.id_map, self.limits)
.and_then(|()| materialize_destinations(&stage, &request.destinations, &request.id_map))
@ -229,9 +275,11 @@ impl PreparedLowerStore {
{
let _ = self.catalog.remove_prepare(&owned);
}
return Err(error);
return Err(PreparedLowerError::materialization(error));
}
self.publish_owned(&owned, &identities)
let physical_lock = self.physical_lock()?;
self.publish_owned(&owned, &identities, &physical_lock)
.map_err(PreparedLowerError::publication)
}
/// Recovers one exact durable prepare operation without adopting foreign paths.
@ -240,7 +288,15 @@ impl PreparedLowerStore {
///
/// Refuses unexpected paths, missing ownership evidence, identity mismatch, and catalog or
/// filesystem failures.
pub fn recover_prepare(&self, operation: &PrepareOperation) -> Result<PrepareRecovery> {
pub fn recover_prepare(
&self,
operation: &PrepareOperation,
) -> PreparedLowerResult<PrepareRecovery> {
self.recover_prepare_inner(operation)
.map_err(PreparedLowerError::recovery)
}
fn recover_prepare_inner(&self, operation: &PrepareOperation) -> Result<PrepareRecovery> {
let expected_stage = stage_path(operation.operation_id);
if operation.stage_path.as_bytes() != path_bytes(&expected_stage) {
bail!("prepare operation stage path is not canonical");
@ -281,12 +337,19 @@ impl PreparedLowerStore {
/// # Errors
/// Stops at the first foreign, mismatched, unreadable, or changed operation without treating
/// it as owned absence.
pub fn recover_preparations(&self) -> Result<Vec<(PrepareOperationId, PrepareRecovery)>> {
pub fn recover_preparations(
&self,
) -> PreparedLowerResult<Vec<(PrepareOperationId, PrepareRecovery)>> {
self.recover_preparations_inner()
.map_err(PreparedLowerError::recovery)
}
fn recover_preparations_inner(&self) -> Result<Vec<(PrepareOperationId, PrepareRecovery)>> {
let operations = self.catalog.snapshot()?.prepare_operations;
operations
.iter()
.map(|operation| {
self.recover_prepare(operation)
self.recover_prepare_inner(operation)
.map(|outcome| (operation.operation_id, outcome))
})
.collect()
@ -300,19 +363,32 @@ impl PreparedLowerStore {
pub fn gc_artifacts(
&self,
selected: &[PreparedArtifactId],
) -> Result<Vec<PreparedArtifactGcResult>> {
) -> PreparedLowerResult<Vec<PreparedArtifactGcResult>> {
self.gc_artifacts_inner(selected)
}
fn gc_artifacts_inner(
&self,
selected: &[PreparedArtifactId],
) -> PreparedLowerResult<Vec<PreparedArtifactGcResult>> {
let mut unique = std::collections::HashSet::new();
if selected.iter().any(|id| !unique.insert(*id)) {
bail!("prepared-artifact GC target is duplicated");
return Err(PreparedLowerError::reclaim(anyhow::anyhow!(
"prepared-artifact GC target is duplicated"
)));
}
let snapshot = self.catalog.snapshot()?;
let snapshot = self
.catalog
.snapshot()
.map_err(PreparedLowerError::catalog)?;
let mut results = Vec::with_capacity(selected.len());
for id in selected {
let artifact = snapshot
.prepared_artifacts
.iter()
.find(|artifact| artifact.prepared_artifact_id == *id)
.context("prepared-artifact GC target is absent")?;
.context("prepared-artifact GC target is absent")
.map_err(PreparedLowerError::reclaim)?;
let outcome = match artifact.state {
ArtifactState::Available => self.reclaim(artifact)?,
ArtifactState::Reclaiming => self.resume_reclaim(artifact)?,
@ -331,8 +407,11 @@ impl PreparedLowerStore {
///
/// Refuses leases, forged catalog rows, physical identity mismatch, foreign paths, and
/// filesystem or catalog failures.
pub fn reclaim(&self, expected: &PreparedArtifact) -> Result<ReclaimOutcome> {
let reclaiming = self.catalog.begin_reclaim(expected)?;
pub fn reclaim(&self, expected: &PreparedArtifact) -> PreparedLowerResult<ReclaimOutcome> {
let reclaiming = self
.catalog
.begin_reclaim(expected)
.map_err(PreparedLowerError::catalog)?;
self.resume_reclaim(&reclaiming)
}
@ -342,7 +421,16 @@ impl PreparedLowerStore {
///
/// Refuses non-reclaiming state, path/identity mismatch, missing physical evidence, and
/// catalog or filesystem failures.
pub fn resume_reclaim(&self, reclaiming: &PreparedArtifact) -> Result<ReclaimOutcome> {
pub fn resume_reclaim(
&self,
reclaiming: &PreparedArtifact,
) -> PreparedLowerResult<ReclaimOutcome> {
let _physical_lock = self.physical_lock()?;
self.resume_reclaim_inner(reclaiming)
.map_err(PreparedLowerError::reclaim)
}
fn resume_reclaim_inner(&self, reclaiming: &PreparedArtifact) -> Result<ReclaimOutcome> {
if reclaiming.state != ArtifactState::Reclaiming {
bail!("artifact is not reclaiming");
}
@ -352,7 +440,6 @@ impl PreparedLowerStore {
}
let quarantine = quarantine_path(reclaiming);
{
let _lock = self.physical_lock()?;
if let Some(canonical) = self.open_relative(&expected_artifact)? {
if file_identity(&canonical)? == (reclaiming.device, reclaiming.inode) {
self.rename_relative_noreplace(&expected_artifact, &quarantine)?;
@ -395,10 +482,10 @@ impl PreparedLowerStore {
&self,
operation: &PrepareOperation,
identities: &Identities,
_physical_lock: &ExclusiveFileLock,
) -> Result<PreparedArtifact> {
let stage = stage_path(operation.operation_id);
let artifact_path = artifact_path(operation.expected_artifact_id);
let _lock = self.physical_lock()?;
if let Some(existing) = self.open_relative(&artifact_path)? {
if let Some(row) = self.available_artifact(operation.expected_artifact_id)? {
identities.require_equivalent(&row)?;
@ -526,20 +613,22 @@ impl PreparedLowerStore {
Ok(())
}
fn physical_lock(&self) -> Result<ExclusiveFileLock> {
fn physical_lock(&self) -> PreparedLowerResult<ExclusiveFileLock> {
let deadline = Instant::now() + PHYSICAL_LOCK_WAIT;
loop {
let file = open_lock_at(&self.root)?;
let file = open_lock_at(&self.root).map_err(PreparedLowerError::locking)?;
match try_exclusive_flock(file) {
Ok(lock) => return Ok(lock),
Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
let now = Instant::now();
if now >= deadline {
bail!("prepared-lower physical lock wait expired");
return Err(PreparedLowerError::locking(anyhow::anyhow!(
"prepared-lower physical lock wait expired"
)));
}
thread::sleep(LOCK_RETRY.min(deadline.saturating_duration_since(now)));
}
Err(error) => return Err(error.into()),
Err(error) => return Err(PreparedLowerError::locking(error)),
}
}
}

343
crates/lazaret-prepared-lower/src/transport.rs
View file

@ -10,6 +10,7 @@ use anyhow::{Context, Result, bail};
use lazaret_core::configuration::ConfinedPathPlan;
use oci_client::Reference;
use oci_client::client::{Client, ClientConfig};
use oci_client::errors::{OciDistributionError, OciErrorCode};
use oci_client::manifest::{IMAGE_LAYER_GZIP_MEDIA_TYPE, IMAGE_LAYER_MEDIA_TYPE, OciImageManifest};
use oci_client::secrets::RegistryAuth;
use rustix::fs::{
@ -19,6 +20,8 @@ use rustix::fs::{
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use crate::{PreparedLowerError, PreparedLowerResult};
const OCI_LAYER_ZSTD: &str = "application/vnd.oci.image.layer.v1.tar+zstd";
const MAX_CONFIG_BYTES: usize = 16 * 1024 * 1024;
const MAX_LAYER_BYTES: u64 = 32 * 1024 * 1024 * 1024;
@ -97,11 +100,15 @@ impl TransportedImage {
///
/// # Errors
/// Refuses a replacement path and reports any descriptor-relative cleanup failure.
pub fn cleanup(mut self) -> Result<()> {
if let Some(layout) = &mut self.retained_layout {
cleanup_retained_layout(layout)?;
pub fn cleanup(mut self) -> PreparedLowerResult<()> {
self.cleanup_inner()
.map_err(PreparedLowerError::publication)
}
Ok(())
fn cleanup_inner(&mut self) -> Result<()> {
self.retained_layout
.as_mut()
.map_or(Ok(()), cleanup_retained_layout)
}
}
@ -127,7 +134,7 @@ impl OciRegistryTransport {
credentials: &RegistryCredentials,
platform: &Platform,
layout_path: &Path,
) -> Result<TransportedImage> {
) -> PreparedLowerResult<TransportedImage> {
self.pull_with_config(
reference,
credentials,
@ -150,7 +157,7 @@ impl OciRegistryTransport {
platform: &Platform,
storage_root: &ConfinedPathPlan,
staging_id: [u8; 32],
) -> Result<TransportedImage> {
) -> PreparedLowerResult<TransportedImage> {
self.pull_to(
reference,
credentials,
@ -171,7 +178,7 @@ impl OciRegistryTransport {
platform: &Platform,
layout_path: &Path,
config: ClientConfig,
) -> Result<TransportedImage> {
) -> PreparedLowerResult<TransportedImage> {
self.pull_to(
reference,
credentials,
@ -189,9 +196,11 @@ impl OciRegistryTransport {
platform: &Platform,
destination: LayoutDestination<'_>,
mut config: ClientConfig,
) -> Result<TransportedImage> {
validate_credentials(credentials)?;
let reference = Reference::from_str(reference).context("invalid OCI image reference")?;
) -> PreparedLowerResult<TransportedImage> {
validate_credentials(credentials).map_err(PreparedLowerError::authentication)?;
let reference = Reference::from_str(reference)
.context("invalid OCI image reference")
.map_err(PreparedLowerError::transport)?;
let selected_platform = platform.clone();
config.max_concurrent_download = 1;
config.platform_resolver = Some(Box::new(move |entries| {
@ -217,54 +226,37 @@ impl OciRegistryTransport {
let (manifest, manifest_digest, config_json) = client
.pull_manifest_and_config(&reference, &auth)
.await
.context("pulling selected OCI manifest and config")?;
.map_err(|error| {
distribution_error(error, "pulling selected OCI manifest and config")
})?;
if config_json.len() > MAX_CONFIG_BYTES {
bail!("OCI config exceeds {MAX_CONFIG_BYTES} bytes");
return Err(PreparedLowerError::manifest_selection(anyhow::anyhow!(
"OCI config exceeds {MAX_CONFIG_BYTES} bytes"
)));
}
verify_platform(&config_json, platform)?;
validate_manifest(&manifest)?;
let layout = LayoutWriter::create(destination).context("creating private OCI layout")?;
verify_platform(&config_json, platform).map_err(PreparedLowerError::platform_selection)?;
validate_manifest(&manifest).map_err(PreparedLowerError::manifest_selection)?;
let layout = LayoutWriter::create(destination)
.context("creating private OCI layout")
.map_err(PreparedLowerError::confinement)?;
layout
.write_verified_blob(&manifest.config.digest, config_json.as_bytes())
.context("writing verified OCI config")?;
for descriptor in &manifest.layers {
let size = u64::try_from(descriptor.size).context("negative OCI layer size")?;
if size > MAX_LAYER_BYTES {
bail!("OCI layer exceeds {MAX_LAYER_BYTES} bytes");
}
let digest_component = digest_component(&descriptor.digest)?;
if layout.blob_exists(digest_component)? {
layout.require_existing_blob(&descriptor.digest, size)?;
continue;
}
let (file, temporary) = layout
.create_blob(digest_component)
.with_context(|| format!("creating OCI layer {}", descriptor.digest))?;
let file = tokio::fs::File::from_std(file);
if let Err(error) = client.pull_blob(&reference, descriptor, file).await {
layout.remove_temporary_blob(&temporary);
return Err(error)
.with_context(|| format!("pulling OCI layer {}", descriptor.digest));
}
let observed = layout
.temporary_blob_size(&temporary)
.with_context(|| format!("measuring OCI layer {}", descriptor.digest))?;
if observed != size {
bail!("OCI layer {} size mismatch", descriptor.digest);
}
layout
.publish_blob(&temporary, digest_component)
.with_context(|| format!("publishing OCI layer {}", descriptor.digest))?;
}
let local_manifest = serde_json::to_vec(&manifest)?;
.context("writing verified OCI config")
.map_err(PreparedLowerError::publication)?;
pull_layers(&client, &reference, &manifest, &layout).await?;
let local_manifest =
serde_json::to_vec(&manifest).map_err(PreparedLowerError::manifest_selection)?;
let local_digest = format!("sha256:{}", hex(&Sha256::digest(&local_manifest)));
layout
.write_verified_blob(&local_digest, &local_manifest)
.context("writing verified OCI manifest")?;
.context("writing verified OCI manifest")
.map_err(PreparedLowerError::publication)?;
layout
.write_layout_index(&local_digest, local_manifest.len())
.context("writing OCI layout index")?;
let selected_manifest_digest = parse_sha256(&local_digest)?;
.context("writing OCI layout index")
.map_err(PreparedLowerError::publication)?;
let selected_manifest_digest =
parse_sha256(&local_digest).map_err(PreparedLowerError::manifest_selection)?;
let source_identity = format!(
"oci-v1\0{}\0{}\0{manifest_digest}",
reference.registry(),
@ -280,6 +272,115 @@ impl OciRegistryTransport {
}
}
async fn pull_layers(
client: &Client,
reference: &Reference,
manifest: &OciImageManifest,
layout: &LayoutWriter,
) -> PreparedLowerResult<()> {
for descriptor in &manifest.layers {
let size = u64::try_from(descriptor.size)
.context("negative OCI layer size")
.map_err(PreparedLowerError::manifest_selection)?;
if size > MAX_LAYER_BYTES {
return Err(PreparedLowerError::layer_validation(anyhow::anyhow!(
"OCI layer exceeds {MAX_LAYER_BYTES} bytes"
)));
}
let digest_component =
digest_component(&descriptor.digest).map_err(PreparedLowerError::manifest_selection)?;
if layout
.blob_exists(digest_component)
.map_err(PreparedLowerError::operating_system)?
{
layout
.require_existing_blob(&descriptor.digest, size)
.map_err(PreparedLowerError::layer_validation)?;
continue;
}
let (file, temporary) = layout
.create_blob(digest_component)
.with_context(|| format!("creating OCI layer {}", descriptor.digest))
.map_err(PreparedLowerError::materialization)?;
let file = tokio::fs::File::from_std(file);
if let Err(error) = client.pull_blob(reference, descriptor, file).await {
layout.remove_temporary_blob(&temporary);
return Err(distribution_error(
error,
format!("pulling OCI layer {}", descriptor.digest),
));
}
let observed = layout
.temporary_blob_size(&temporary)
.with_context(|| format!("measuring OCI layer {}", descriptor.digest))
.map_err(PreparedLowerError::operating_system)?;
if observed != size {
return Err(PreparedLowerError::layer_validation(anyhow::anyhow!(
"OCI layer {} size mismatch",
descriptor.digest
)));
}
layout
.publish_blob(&temporary, digest_component)
.with_context(|| format!("publishing OCI layer {}", descriptor.digest))
.map_err(PreparedLowerError::publication)?;
}
Ok(())
}
fn distribution_error(
error: OciDistributionError,
context: impl Into<String>,
) -> PreparedLowerError {
let authentication = matches!(
&error,
OciDistributionError::AuthenticationFailure(_)
| OciDistributionError::RegistryTokenDecodeError(_)
| OciDistributionError::UnauthorizedError { .. }
| OciDistributionError::ServerError {
code: 401 | 403,
..
}
) || matches!(
&error,
OciDistributionError::RegistryError { envelope, .. }
if envelope.errors.iter().any(|error| {
matches!(error.code, OciErrorCode::Unauthorized | OciErrorCode::Denied)
})
) || matches!(
&error,
OciDistributionError::RequestError(error)
if error.status().is_some_and(|status| matches!(status.as_u16(), 401 | 403))
);
let platform = matches!(
&error,
OciDistributionError::ImageIndexParsingNoPlatformResolverError
);
let manifest = matches!(
&error,
OciDistributionError::ConfigConversionError(_)
| OciDistributionError::ImageManifestNotFoundError(_)
| OciDistributionError::IncompatibleLayerMediaTypeError(_)
| OciDistributionError::JsonError(_)
| OciDistributionError::ManifestEncodingError(_)
| OciDistributionError::ManifestParsingError(_)
| OciDistributionError::PullNoLayersError
| OciDistributionError::UnsupportedMediaTypeError(_)
| OciDistributionError::UnsupportedSchemaVersionError(_)
| OciDistributionError::VersionedParsingError(_)
);
let source = anyhow::Error::new(error).context(context.into());
if authentication {
PreparedLowerError::authentication(source)
} else if platform {
PreparedLowerError::platform_selection(source)
} else if manifest {
PreparedLowerError::manifest_selection(source)
} else {
PreparedLowerError::transport(source)
}
}
#[derive(Deserialize)]
struct ImageConfigPlatform {
os: String,
@ -726,146 +827,4 @@ fn hex(bytes: &[u8]) -> String {
)
}
#[cfg(test)]
mod tests {
use std::io::{Read, Write};
use std::net::TcpListener;
use std::os::unix::fs::PermissionsExt;
use oci_client::client::ClientProtocol;
use super::*;
#[test]
fn atomic_layout_write_accepts_a_root_level_destination() {
let tree = tempfile::tempdir().unwrap();
let root = File::open(tree.path()).unwrap();
write_relative_atomic(&root, Path::new("oci-layout"), b"layout").unwrap();
assert_eq!(
std::fs::read(tree.path().join("oci-layout")).unwrap(),
b"layout"
);
}
#[test]
fn confined_read_open_uses_an_empty_creation_mode() {
let tree = tempfile::tempdir().unwrap();
let root = File::open(tree.path()).unwrap();
let error = open_relative_file(&root, Path::new("absent"), OFlags::RDONLY).unwrap_err();
assert_eq!(error.kind(), std::io::ErrorKind::NotFound);
}
#[test]
fn preexisting_transport_parents_require_exact_private_mode() {
for name in ["image-transports", "blobs", "sha256"] {
let tree = tempfile::tempdir().unwrap();
std::fs::create_dir(tree.path().join(name)).unwrap();
std::fs::set_permissions(
tree.path().join(name),
std::fs::Permissions::from_mode(0o777),
)
.unwrap();
let parent = File::open(tree.path()).unwrap();
assert!(
create_directory(&parent, name).is_err(),
"accepted wrong mode for {name}"
);
}
}
#[test]
fn refuses_mismatched_and_incomplete_platforms() {
assert!(
verify_platform(
r#"{"os":"linux","architecture":"amd64"}"#,
&Platform::linux("amd64")
)
.is_ok()
);
assert!(
verify_platform(
r#"{"os":"linux","architecture":"arm64"}"#,
&Platform::linux("amd64")
)
.is_err()
);
assert!(
verify_platform(
r#"{"os":"linux","architecture":"amd64","variant":"v2"}"#,
&Platform::linux("amd64")
)
.is_err()
);
assert!(verify_platform(r#"{"architecture":"amd64"}"#, &Platform::linux("amd64")).is_err());
}
#[test]
fn incomplete_authentication_refuses_without_layout_or_secret_disclosure() {
let runtime = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
let temp = tempfile::tempdir().unwrap();
let result = runtime.block_on(OciRegistryTransport.pull_with_config(
"registry.invalid/private/image:latest",
&RegistryCredentials::Basic {
username: "developer".to_owned(),
password: String::new(),
},
&Platform::linux("amd64"),
&temp.path().join("layout"),
ClientConfig::default(),
));
let error = result.unwrap_err().to_string();
assert!(error.contains("credentials are incomplete"));
assert!(!error.contains("developer"));
assert!(!temp.path().join("layout").exists());
assert_eq!(
format!(
"{:?}",
RegistryCredentials::Bearer("not-a-real-token".to_owned())
),
"Bearer(\"[redacted]\")"
);
}
#[test]
fn registry_authentication_failure_refuses_before_layout_creation() {
let listener = match TcpListener::bind("127.0.0.1:0") {
Ok(listener) => listener,
Err(error) if error.kind() == std::io::ErrorKind::PermissionDenied => return,
Err(error) => panic!("binding test registry: {error}"),
};
let address = listener.local_addr().unwrap();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut request = [0; 4096];
assert!(stream.read(&mut request).unwrap() > 0);
stream
.write_all(
b"HTTP/1.1 401 Unauthorized\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
)
.unwrap();
});
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let temp = tempfile::tempdir().unwrap();
let result = runtime.block_on(OciRegistryTransport.pull_with_config(
&format!("{address}/private/image:latest"),
&RegistryCredentials::Basic {
username: "wrong".to_owned(),
password: "wrong".to_owned(),
},
&Platform::linux("amd64"),
&temp.path().join("layout"),
ClientConfig {
protocol: ClientProtocol::Http,
..ClientConfig::default()
},
));
assert!(result.is_err());
assert!(!temp.path().join("layout").exists());
server.join().unwrap();
}
}
include!("transport/tests.rs");

193
crates/lazaret-prepared-lower/src/transport/tests.rs Normal file
View file

@ -0,0 +1,193 @@
#[cfg(test)]
mod tests {
use std::io::{Read, Write};
use std::net::TcpListener;
use std::os::unix::fs::PermissionsExt;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use oci_client::client::ClientProtocol;
use super::*;
#[test]
fn atomic_layout_write_accepts_a_root_level_destination() {
let tree = tempfile::tempdir().unwrap();
let root = File::open(tree.path()).unwrap();
write_relative_atomic(&root, Path::new("oci-layout"), b"layout").unwrap();
assert_eq!(
std::fs::read(tree.path().join("oci-layout")).unwrap(),
b"layout"
);
}
#[test]
fn confined_read_open_uses_an_empty_creation_mode() {
let tree = tempfile::tempdir().unwrap();
let root = File::open(tree.path()).unwrap();
let error = open_relative_file(&root, Path::new("absent"), OFlags::RDONLY).unwrap_err();
assert_eq!(error.kind(), std::io::ErrorKind::NotFound);
}
#[test]
fn preexisting_transport_parents_require_exact_private_mode() {
for name in ["image-transports", "blobs", "sha256"] {
let tree = tempfile::tempdir().unwrap();
std::fs::create_dir(tree.path().join(name)).unwrap();
std::fs::set_permissions(
tree.path().join(name),
std::fs::Permissions::from_mode(0o777),
)
.unwrap();
let parent = File::open(tree.path()).unwrap();
assert!(
create_directory(&parent, name).is_err(),
"accepted wrong mode for {name}"
);
}
}
#[test]
fn refuses_mismatched_and_incomplete_platforms() {
assert!(
verify_platform(
r#"{"os":"linux","architecture":"amd64"}"#,
&Platform::linux("amd64")
)
.is_ok()
);
assert!(
verify_platform(
r#"{"os":"linux","architecture":"arm64"}"#,
&Platform::linux("amd64")
)
.is_err()
);
assert!(
verify_platform(
r#"{"os":"linux","architecture":"amd64","variant":"v2"}"#,
&Platform::linux("amd64")
)
.is_err()
);
assert!(verify_platform(r#"{"architecture":"amd64"}"#, &Platform::linux("amd64")).is_err());
}
#[test]
fn incomplete_authentication_refuses_without_layout_or_secret_disclosure() {
let runtime = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
let temp = tempfile::tempdir().unwrap();
let result = runtime.block_on(OciRegistryTransport.pull_with_config(
"registry.invalid/private/image:latest",
&RegistryCredentials::Basic {
username: "developer".to_owned(),
password: String::new(),
},
&Platform::linux("amd64"),
&temp.path().join("layout"),
ClientConfig::default(),
));
let error = result.unwrap_err();
assert!(matches!(error, PreparedLowerError::Authentication { .. }));
let error = error.to_string();
assert!(error.contains("credentials are incomplete"));
assert!(!error.contains("developer"));
assert!(!temp.path().join("layout").exists());
assert_eq!(
format!(
"{:?}",
RegistryCredentials::Bearer("not-a-real-token".to_owned())
),
"Bearer(\"[redacted]\")"
);
}
#[test]
fn registry_authentication_failure_refuses_before_layout_creation() {
let listener = match TcpListener::bind("127.0.0.1:0") {
Ok(listener) => listener,
Err(error) if error.kind() == std::io::ErrorKind::PermissionDenied => return,
Err(error) => panic!("binding test registry: {error}"),
};
let address = listener.local_addr().unwrap();
listener.set_nonblocking(true).unwrap();
let stop = Arc::new(AtomicBool::new(false));
let server_stop = Arc::clone(&stop);
let server = std::thread::spawn(move || {
while !server_stop.load(Ordering::Relaxed) {
match listener.accept() {
Ok((mut stream, _)) => {
let mut request = [0; 4096];
assert!(stream.read(&mut request).unwrap() > 0);
stream
.write_all(
b"HTTP/1.1 401 Unauthorized\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
)
.unwrap();
}
Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
std::thread::sleep(Duration::from_millis(1));
}
Err(error) => panic!("accepting test registry request: {error}"),
}
}
});
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let temp = tempfile::tempdir().unwrap();
let result = runtime.block_on(OciRegistryTransport.pull_with_config(
&format!("{address}/private/image:latest"),
&RegistryCredentials::Basic {
username: "wrong".to_owned(),
password: "wrong".to_owned(),
},
&Platform::linux("amd64"),
&temp.path().join("layout"),
ClientConfig {
protocol: ClientProtocol::Http,
..ClientConfig::default()
},
));
assert!(
matches!(result, Err(PreparedLowerError::Authentication { .. })),
"{result:?}"
);
assert!(!temp.path().join("layout").exists());
stop.store(true, Ordering::Relaxed);
server.join().unwrap();
}
#[test]
fn registry_failures_map_to_stable_caller_categories() {
assert!(matches!(
distribution_error(
OciDistributionError::AuthenticationFailure("denied".into()),
"pull"
),
PreparedLowerError::Authentication { .. }
));
assert!(matches!(
distribution_error(
OciDistributionError::ImageIndexParsingNoPlatformResolverError,
"pull"
),
PreparedLowerError::PlatformSelection { .. }
));
assert!(matches!(
distribution_error(
OciDistributionError::ManifestParsingError("invalid".into()),
"pull"
),
PreparedLowerError::ManifestSelection { .. }
));
assert!(matches!(
distribution_error(OciDistributionError::GenericError(None), "pull"),
PreparedLowerError::Transport { .. }
));
}
}

7
crates/lazaret-prepared-lower/tests/prepared_lower.rs
View file

@ -15,7 +15,7 @@ use lazaret_core::identity::{
use lazaret_core::records::{PrepareOperation, PrepareStageState};
use lazaret_prepared_lower::{
IdMap, IdMapRange, PrepareRecovery, PrepareRequest, PreparedDestination,
PreparedDestinationKind, PreparedLowerStore, ReclaimOutcome,
PreparedDestinationKind, PreparedLowerError, PreparedLowerStore, ReclaimOutcome,
};
use lazaret_storage_catalog::{Catalog, CatalogError};
use serde_json::json;
@ -81,7 +81,10 @@ fn preexisting_fixed_children_require_exact_private_mode() {
)
.unwrap();
assert!(
PreparedLowerStore::initialize(&fixture.root, fixture.catalog.clone()).is_err(),
matches!(
PreparedLowerStore::initialize(&fixture.root, fixture.catalog.clone()),
Err(PreparedLowerError::Confinement { .. })
),
"accepted wrong mode for {name}"
);
}

12
crates/lazaret-prepared-lower/tests/prepared_lower/publication.rs
View file

@ -279,7 +279,10 @@ fn recovery_deletes_only_exact_owned_stage_and_preserves_foreign_stage() {
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!(matches!(
fixture.store.recover_prepare(&intent),
Err(PreparedLowerError::Recovery { .. })
));
assert!(fixture.root.join(&stage).join("foreign").exists());
std::fs::remove_dir_all(fixture.root.join(&stage)).unwrap();
assert_eq!(
@ -337,7 +340,7 @@ fn reclaim_refuses_forged_identity_and_leases() {
fixture.catalog.acquire_lease(&lease).unwrap();
assert!(matches!(
fixture.store.reclaim(&artifact),
Err(error)
Err(error @ PreparedLowerError::Catalog { .. })
if matches!(error.downcast_ref::<CatalogError>(), Some(CatalogError::LeasesPresent))
));
fixture.catalog.release_lease(&lease).unwrap();
@ -367,7 +370,10 @@ fn selected_gc_is_idempotent_only_at_the_durable_reclaim_boundary() {
))
.unwrap();
let id = artifact.prepared_artifact_id;
assert!(fixture.store.gc_artifacts(&[id, id]).is_err());
assert!(matches!(
fixture.store.gc_artifacts(&[id, id]),
Err(PreparedLowerError::Reclaim { .. })
));
let lease = lazaret_core::records::ArtifactLease {
prepared_artifact_id: id,

20
crates/lazaret-prepared-lower/tests/prepared_lower/semantics.rs
View file

@ -69,7 +69,10 @@ fn request_manifest_identity_must_match_verified_layout_manifest() {
let mut request = fixture.request(3, &[(bytes, OCI_TAR)]);
request.selected_manifest_digest = [0xff; 32];
assert!(fixture.store.prepare(&request).is_err());
assert!(matches!(
fixture.store.prepare(&request),
Err(PreparedLowerError::ManifestSelection { .. })
));
let snapshot = fixture.catalog.snapshot().unwrap();
assert!(snapshot.prepared_artifacts.is_empty());
assert!(snapshot.prepare_operations.is_empty());
@ -191,7 +194,10 @@ fn unmapped_nonzero_owners_refuse_before_publication() {
}],
};
assert!(fixture.store.prepare(&request).is_err());
assert!(matches!(
fixture.store.prepare(&request),
Err(PreparedLowerError::Materialization { .. })
));
assert!(
fixture
.catalog
@ -258,7 +264,10 @@ fn hostile_and_corrupt_layers_refuse_without_publication() {
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());
assert!(matches!(
fixture.store.prepare(&request),
Err(PreparedLowerError::Materialization { .. })
));
let snapshot = fixture.catalog.snapshot().unwrap();
assert!(snapshot.prepared_artifacts.is_empty());
assert!(snapshot.prepare_operations.is_empty());
@ -291,7 +300,10 @@ fn hostile_and_corrupt_layers_refuse_without_publication() {
.find(|path| std::fs::metadata(path).unwrap().len() > 1024)
.unwrap();
OpenOptionsExt::append_byte(&blob);
assert!(fixture.store.prepare(&request).is_err());
assert!(matches!(
fixture.store.prepare(&request),
Err(PreparedLowerError::LayerValidation { .. })
));
assert!(
fixture
.catalog